Merge -c 1477900 from branch-1 to branch-1.2 to fix MAPREDUCE-5154. Ensure job delegation token renewal is cancelled after job staging directory is deleted. Contributed by Sandy Ryza & Arun C. Murthy.

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

CHANGES.txt

@@ -605,6 +605,9 @@ Release 1.2.0 - 2013.04.16
     causing failures since task directories were being cleaned up in multiple
     threads. (arpit via acmurthy)
 
+    MAPREDUCE-5154. Ensure job delegation token renewal is cancelled after job
+    staging directory is deleted. (Sandy Ryza & Arun C. Murthy via acmurthy)
+
 Release 1.1.2 - 2013.01.30
 
   INCOMPATIBLE CHANGES

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

@@ -27,6 +27,7 @@ import org.apache.commons.logging.LogFactory;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.mapreduce.security.token.DelegationTokenRenewal;
 import org.apache.hadoop.security.UserGroupInformation;
 
 public class CleanupQueue {
@@ -57,16 +58,38 @@ public class CleanupQueue {
     final Path fullPath;// full path of file or dir
     final Configuration conf;
     final UserGroupInformation ugi;
+    final JobID jobIdTokenRenewalToCancel;
 
     public PathDeletionContext(Path fullPath, Configuration conf) {
-      this(fullPath, conf, null);
+      this(fullPath, conf, null, null);
     }
 
     public PathDeletionContext(Path fullPath, Configuration conf,
         UserGroupInformation ugi) {
+      this(fullPath, conf, ugi, null);
+    }
+    
+    /**
+     * PathDeletionContext ctor which also allows for a job-delegation token
+     * renewal to be cancelled.
+     * 
+     * This is usually used at the end of a job to delete it's final path and 
+     * to cancel renewal of it's job-delegation token.
+     * 
+     * @param fullPath path to be deleted
+     * @param conf job configuration
+     * @param ugi ugi of the job to be used to delete the path
+     * @param jobIdTokenRenewalToCancel jobId of the job whose job-delegation
+     *                                  token renewal should be cancelled. No
+     *                                  cancellation is attempted if this is
+     *                                  <code>null</code>
+     */
+    public PathDeletionContext(Path fullPath, Configuration conf,
+        UserGroupInformation ugi, JobID jobIdTokenRenewalToCancel) {
       this.fullPath = fullPath;
       this.conf = conf;
       this.ugi = ugi;
+      this.jobIdTokenRenewalToCancel = jobIdTokenRenewalToCancel;
     }
     
     protected Path getPathForCleanup() {
@@ -86,6 +109,13 @@ public class CleanupQueue {
              return null;
             }
           });
+      
+      // Cancel renewal of job-delegation token if necessary
+      if (jobIdTokenRenewalToCancel != null && 
+          conf.getBoolean(JobContext.JOB_CANCEL_DELEGATION_TOKEN, true)) {
+        DelegationTokenRenewal.removeDelegationTokenRenewalForJob(
+            jobIdTokenRenewalToCancel);
+      }
     }
 
     @Override

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

@@ -3319,13 +3319,13 @@ public class JobInProgress {
         CleanupQueue.getInstance().addToQueue(
             new PathDeletionContext(tempDir, conf));
 
-        // delete the staging area for the job
+        // delete the staging area for the job and cancel delegation token
         String jobTempDir = conf.get("mapreduce.job.dir");
         if (jobTempDir != null && conf.getKeepTaskFilesPattern() == null &&
             !conf.getKeepFailedTaskFiles()) {
           Path jobTempDirPath = new Path(jobTempDir);
           CleanupQueue.getInstance().addToQueue(
-              new PathDeletionContext(jobTempDirPath, conf, userUGI));
+              new PathDeletionContext(jobTempDirPath, conf, userUGI, jobId));
         }
 
       } catch (IOException e) {
@@ -3341,11 +3341,6 @@ public class JobInProgress {
       this.runningReduces = null;
     }
     
-    // remove jobs delegation tokens
-    if(conf.getBoolean(JobContext.JOB_CANCEL_DELEGATION_TOKEN, true)) {
-      DelegationTokenRenewal.removeDelegationTokenRenewalForJob(jobId);
-    } // else don't remove it.May be used by spawned tasks
-
     //close the user's FS
     try {
       fs.close();