Merge -r 677263:677264 from trunk onto 0.18 branch. Fixes HADOOP-3670.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/core/branches/branch-0.18@677267 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -755,6 +755,9 @@ Release 0.18.0 - Unreleased
     HADOOP-3737. Fix CompressedWritable to call Deflater::end to release
     compressor memory. (Grant Glouser via cdouglas)
 
+    HADOOP-3670. Fixes JobTracker to clear out split bytes when no longer 
+    required. (Amareshwari Sriramadasu via ddas)
+
 Release 0.17.2 - Unreleased
 
   BUG FIXES

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

@@ -757,6 +757,10 @@ public class JobClient extends Configured implements MRConstants, Tool  {
     public BytesWritable getBytes() {
       return bytes;
     }
+
+    public void clearBytes() {
+      bytes = null;
+    }
       
     public void setLocations(String[] locations) {
       this.locations = locations;

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

@@ -1638,6 +1638,11 @@ class JobInProgress {
         localJarFile = null;
       }
 
+      // clean up splits
+      for (int i = 0; i < maps.length; i++) {
+        maps[i].clearSplit();
+      }
+
       // JobClient always creates a new directory with job files
       // so we remove that directory to cleanup
       // Delete temp dfs dirs created if any, like in case of 

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

@@ -86,7 +86,7 @@ class MapTask extends Task {
                  ) throws IOException {
     super(jobFile, taskId, partition);
     this.splitClass = splitClass;
-    this.split.set(split);
+    this.split = split;
   }
 
   @Override
@@ -116,6 +116,7 @@ class MapTask extends Task {
     super.write(out);
     Text.writeString(out, splitClass);
     split.write(out);
+    split = null;
   }
   
   @Override

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

@@ -777,6 +777,9 @@ class TaskInProgress {
    * Gets the Node list of input split locations sorted in rack order.
    */ 
   public String getSplitNodes() {
+    if ( rawSplit == null) {
+      return "";
+    }
     String[] splits = rawSplit.getLocations();
     Node[] nodes = new Node[splits.length];
     for (int i = 0; i < splits.length; i++) {
@@ -805,4 +808,8 @@ class TaskInProgress {
     return ret.toString();
   }
 
+  public void clearSplit() {
+    rawSplit.clearBytes();
+  }
+
 }