Merge -c 1476463 from branch-1 to branch-1.2 to fix MAPREDUCE-5158. Cleanup job-submission related files on HDFS during JobTracker restart for jobs which set mapreduce.job.restart.recover to false to indicate they don't want to be recovered. Contributed by Mayank Bansal.

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

CHANGES.txt

@@ -594,6 +594,11 @@ Release 1.2.0 - 2013.04.16
     HADOOP-9502. chmod/chown do not return error exit codes for some exceptions.
     (szetszwo)
 
+    MAPREDUCE-5158. Cleanup job-submission related files on HDFS during
+    JobTracker restart for jobs which set mapreduce.job.restart.recover to
+    false to indicate they don't want to be recovered. (Mayank Bansal via
+    acmurthy) 
+
 Release 1.1.2 - 2013.01.30
 
   INCOMPATIBLE CHANGES

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

@@ -3298,6 +3298,16 @@ public class JobInProgress {
       jobtracker.storeCompletedJob(this);
       jobtracker.finalizeJob(this);
 
+    }
+    cleanupJob();
+  }
+  
+  /**
+   * The job is dead. We're now cleaning it, getting rid of job directories and
+   * removing all delegation token etc.
+   */
+  void cleanupJob() {
+    synchronized (this) {
       try {
         // Definitely remove the local-disk copy of the job file
         if (localJobFile != null) {

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

@@ -3609,6 +3609,8 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
         !job.getJobConf().getBoolean(
             JobConf.MAPREDUCE_RECOVER_JOB, 
             JobConf.DEFAULT_MAPREDUCE_RECOVER_JOB)) {
+      LOG.info("Job "+ jobId.toString() + " is not enable for recovery, cleaning up job files");
+      job.cleanupJob();
       return null;
     }
     

src/test/org/apache/hadoop/mapred/TestRecoveryManager.java

@@ -257,6 +257,8 @@ public class TestRecoveryManager {
       UtilsForTests.waitFor(100);
     }
     Assert.assertTrue("Task should be successful", rJob1.isSuccessful());
+    Assert.assertTrue("Job should be cleaned up", !fs.exists(new Path(job1.get("mapreduce.job.dir"))));
+    Assert.assertTrue("Job should be cleaned up", !fs.exists(new Path(job2.get("mapreduce.job.dir"))));
   }
 
   public static class TestJobTrackerInstrumentation extends JobTrackerInstrumentation {