MAPREDUCE-5028. Maps fail when io.sort.mb is set to high value. (kkambatl via tucu)

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-1@1453107 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -517,6 +517,9 @@ Release 1.2.0 - unreleased
     HADOOP-9349. Confusing output when running hadoop version from one hadoop 
     installation when HADOOP_HOME points to another. (sandyr via tucu)
 
+    MAPREDUCE-5028. Maps fail when io.sort.mb is set to high value.
+    (kkambatl via tucu)
+
 Release 1.1.2 - Unreleased
 
   INCOMPATIBLE CHANGES

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

@@ -1256,7 +1256,8 @@ abstract public class Task implements Writable, Configurable {
       more = in.next();
       if (more) {
         DataInputBuffer nextKeyBytes = in.getKey();
-        keyIn.reset(nextKeyBytes.getData(), nextKeyBytes.getPosition(), nextKeyBytes.getLength());
+        keyIn.reset(nextKeyBytes.getData(), nextKeyBytes.getPosition(),
+            nextKeyBytes.getLength() - nextKeyBytes.getPosition());
         nextKey = keyDeserializer.deserialize(nextKey);
         hasNext = key != null && (comparator.compare(key, nextKey) == 0);
       } else {
@@ -1270,7 +1271,8 @@ abstract public class Task implements Writable, Configurable {
      */
     private void readNextValue() throws IOException {
       DataInputBuffer nextValueBytes = in.getValue();
-      valueIn.reset(nextValueBytes.getData(), nextValueBytes.getPosition(), nextValueBytes.getLength());
+      valueIn.reset(nextValueBytes.getData(), nextValueBytes.getPosition(),
+          nextValueBytes.getLength() - nextValueBytes.getPosition());
       value = valDeserializer.deserialize(value);
     }
   }

src/mapred/org/apache/hadoop/mapreduce/ReduceContext.java

@@ -112,7 +112,8 @@ public class ReduceContext<KEYIN,VALUEIN,KEYOUT,VALUEOUT>
     buffer.reset(currentRawKey.getBytes(), 0, currentRawKey.getLength());
     key = keyDeserializer.deserialize(key);
     next = input.getValue();
-    buffer.reset(next.getData(), next.getPosition(), next.getLength());
+    buffer.reset(next.getData(), next.getPosition(),
+        next.getLength() - next.getPosition());
     value = valueDeserializer.deserialize(value);
     hasMore = input.next();
     if (hasMore) {