HADOOP-2997. Adds test for non-writable serialier. Also fixes a problem introduced by HADOOP-2399. Contirbuted by Tom White.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/core/trunk@642164 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -398,6 +398,9 @@ Trunk (unreleased changes)
     HADOOP-3064. Commas in a file path should not be treated as delimiters.
     (Hairong Kuang via shv)
 
+    HADOOP-2997. Adds test for non-writable serialier. Also fixes a problem 
+    introduced by HADOOP-2399. (Tom White via ddas)
+
 Release 0.16.2 - Unreleased
 
   BUG FIXES

src/java/org/apache/hadoop/mapred/ReduceTask.java

@@ -209,8 +209,6 @@ class ReduceTask extends Task {
       this.in = in;
       this.comparator = comparator;
       this.reporter = reporter;
-      nextKey = (KEY) ReflectionUtils.newInstance(keyClass, conf);
-      value = (VALUE) ReflectionUtils.newInstance(valClass, conf);
       SerializationFactory serializationFactory = new SerializationFactory(conf);
       this.keyDeserializer = serializationFactory.getDeserializer(keyClass);
       this.keyDeserializer.open(keyIn);
@@ -218,7 +216,7 @@ class ReduceTask extends Task {
       this.valDeserializer.open(valIn);
       readNextKey();
       key = nextKey;
-      nextKey = (KEY) ReflectionUtils.newInstance(keyClass, conf);
+      nextKey = null; // force new instance creation
       hasNext = more;
     }
 
@@ -275,7 +273,7 @@ class ReduceTask extends Task {
       if (more) {
         DataOutputBuffer nextKeyBytes = in.getKey();
         keyIn.reset(nextKeyBytes.getData(), nextKeyBytes.getLength());
-        keyDeserializer.deserialize(nextKey);
+        nextKey = keyDeserializer.deserialize(nextKey);
         hasNext = key != null && (comparator.compare(key, nextKey) == 0);
       } else {
         hasNext = false;
@@ -290,7 +288,7 @@ class ReduceTask extends Task {
       nextValue.reset();
       in.getValue().writeUncompressedBytes(nextValue);
       valIn.reset(nextValue.getData(), nextValue.getLength());
-      valDeserializer.deserialize(value);
+      value = valDeserializer.deserialize(value);
     }
   }