HADOOP-4654. Removes temporary output directory for failed and killed tasks in the JobTracker's task commit thread. Contributed by Amareshwari Sriramadasu.

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

CHANGES.txt

@@ -71,6 +71,10 @@ Release 0.18.3 - Unreleased
 
     HADOOP-4746. Job output directory should be normalized. (hairong)
 
+    HADOOP-4654. Removes temporary output directory for failed and killed
+    tasks in the JobTracker's task commit thread. 
+    (Amareshwari Sriramadasu via ddas)
+
 Release 0.18.2 - 2008-11-03
 
   BUG FIXES

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

@@ -524,6 +524,10 @@ class JobInProgress {
         return;
       } else if (state == TaskStatus.State.FAILED ||
                  state == TaskStatus.State.KILLED) {
+        // To remove the temporary output of failed/killed tasks
+        JobWithTaskContext j = new JobWithTaskContext(this, tip, 
+                                     status.getTaskID(), metrics);
+        jobtracker.addToCommitQueue(j);
         // Get the event number for the (possibly) previously successful
         // task. If there exists one, then set that status to OBSOLETE 
         int eventNumber;

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

@@ -2318,6 +2318,15 @@ public class JobTracker implements MRConstants, InterTrackerProtocol, JobSubmiss
                     isTipComplete[index] = true;
                   }
                 }
+              } else if (states[index] == TaskStatus.State.FAILED || 
+                         states[index] == TaskStatus.State.KILLED) {
+                try {
+                  tasks[index].removeTaskOutput();
+                } catch (IOException e) {
+                  LOG.info("Failed to remove temporary directory of "
+                           + status[index].getTaskID() + " with " 
+                           + StringUtils.stringifyException(e));
+                }
               }
             } catch (IOException ioe) {
               // Oops! Failed to copy the task's output to its final place;

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

@@ -571,6 +571,14 @@ abstract class Task implements Writable, Configurable {
     }
   }
   
+  void removeTaskOutput() throws IOException {
+    if (taskOutputPath != null) {
+      FileSystem fs = taskOutputPath.getFileSystem(conf);
+      // Delete the temporary task-specific output directory
+      fs.delete(taskOutputPath, true);
+    }
+  }
+  
   private Path getFinalPath(Path jobOutputDir, Path taskOutput) {
     URI relativePath = taskOutputPath.toUri().relativize(taskOutput.toUri());
     if (relativePath.getPath().length() > 0) {