MAPREDUCE-5508. JobTracker memory leak caused by unreleased FileSystem objects in JobInProgress#cleanupJob. Contributed by Xi Fang.

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

CHANGES.txt

@@ -139,6 +139,9 @@ Release 1.3.0 - unreleased
     FileSystem#close can cause return of a closed DistributedFileSystem instance
     from the FileSystem cache. (cnauroth)
 
+    MAPREDUCE-5508. JobTracker memory leak caused by unreleased FileSystem
+    objects in JobInProgress#cleanupJob. (Xi Fang via cnauroth)
+
 Release 1.2.2 - unreleased
 
   INCOMPATIBLE CHANGES

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

@@ -60,14 +60,10 @@ public class CleanupQueue {
     final Configuration conf;
     final UserGroupInformation ugi;
     final JobID jobIdTokenRenewalToCancel;
+    FileSystem fs;
 
     public PathDeletionContext(Path fullPath, Configuration conf) {
-      this(fullPath, conf, null, null);
-    }
-
-    public PathDeletionContext(Path fullPath, Configuration conf,
-        UserGroupInformation ugi) {
-      this(fullPath, conf, ugi, null);
+      this(fullPath, conf, null, null, null);
     }
     
     /**
@@ -86,11 +82,12 @@ public class CleanupQueue {
      *                                  <code>null</code>
      */
     public PathDeletionContext(Path fullPath, Configuration conf,
-        UserGroupInformation ugi, JobID jobIdTokenRenewalToCancel) {
+        UserGroupInformation ugi, JobID jobIdTokenRenewalToCancel, FileSystem fs) {
       this.fullPath = fullPath;
       this.conf = conf;
       this.ugi = ugi;
       this.jobIdTokenRenewalToCancel = jobIdTokenRenewalToCancel;
+      this.fs = fs;
     }
     
     protected Path getPathForCleanup() {
@@ -106,7 +103,7 @@ public class CleanupQueue {
       (ugi == null ? UserGroupInformation.getLoginUser() : ugi).doAs(
           new PrivilegedExceptionAction<Object>() {
             public Object run() throws IOException {
-              FileSystem fs = p.getFileSystem(conf);
+              fs = (fs == null ?  p.getFileSystem(conf) : fs);
               try {
                 fs.delete(p, true);
                 return null;

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

@@ -3330,10 +3330,17 @@ public class JobInProgress {
         String jobTempDir = conf.get("mapreduce.job.dir");
         if (jobTempDir != null && conf.getKeepTaskFilesPattern() == null &&
             !conf.getKeepFailedTaskFiles()) {
-          Path jobTempDirPath = new Path(jobTempDir);
-          tempDirFs = jobTempDirPath.getFileSystem(conf);
+          final Path jobTempDirPath = new Path(jobTempDir);
+          try {
+            tempDirFs = userUGI.doAs(new PrivilegedExceptionAction<FileSystem>() {
+              public FileSystem run() throws IOException {
+                return jobTempDirPath.getFileSystem(conf);
+              }});
+          } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+          }
           CleanupQueue.getInstance().addToQueue(
-              new PathDeletionContext(jobTempDirPath, conf, userUGI, jobId));
+              new PathDeletionContext(jobTempDirPath, conf, userUGI, jobId, tempDirFs));
         }
 
       } catch (IOException e) {

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

@@ -44,7 +44,7 @@ public class TestCleanupQueue {
     // With UGI, should close FileSystem
     CleanupQueue cleanupQueue = new CleanupQueue();
     PathDeletionContext context = new PathDeletionContext(path, conf,
-        UserGroupInformation.getLoginUser());
+        UserGroupInformation.getLoginUser(), null, null);
     cleanupQueue.addToQueue(context);
     
     while (FileSystem.getCacheSize() > 0) {