MAPREDUCE-3343. TaskTracker Out of Memory because of distributed cache. Contributed by Zhao Yunjiong

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

CHANGES.txt

@@ -35,6 +35,9 @@ Release 0.20.206.0 - unreleased
     MAPREDUCE-2073. TestTrackerDistributedCacheManager should be
     up-front about requirements on build environment. (todd)
 
+    MAPREDUCE-3343. TaskTracker Out of Memory because of distributed cache.
+    (Zhao Yunjiong).
+
   IMPROVEMENTS
 
     MAPREDUCE-2836. Provide option to fail jobs when submitted to

src/mapred/org/apache/hadoop/filecache/TrackerDistributedCacheManager.java

@@ -675,6 +675,18 @@ public class TrackerDistributedCacheManager {
     }
   }
 
+  public void removeTaskDistributedCacheManager(JobID jobId) {
+    jobArchives.remove(jobId);
+  }
+
+  /*
+   * This method is called from unit tests.
+   */
+  protected TaskDistributedCacheManager getTaskDistributedCacheManager(
+      JobID jobId) {
+    return jobArchives.get(jobId);
+  }
+
   /**
    * Determines timestamps of files to be cached, and stores those
    * in the configuration.  This is intended to be used internally by JobClient

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

@@ -2081,8 +2081,9 @@ public class TaskTracker implements MRConstants, TaskUmbilicalProtocol,
       runningJobs.remove(jobId);
     }
     getJobTokenSecretManager().removeTokenForJob(jobId.toString());  
-  }      
-    
+    distributedCacheManager.removeTaskDistributedCacheManager(jobId);
+  }
+
   /**
    * This job's files are no longer needed on this TT, remove them.
    *

src/test/org/apache/hadoop/filecache/TestTrackerDistributedCacheManager.java

@@ -1147,4 +1147,23 @@ public class TestTrackerDistributedCacheManager extends TestCase {
     }
   }
 
+  public void testRemoveTaskDistributedCacheManager() throws Exception {
+    if (!canRun()) {
+      return;
+    }
+    TrackerDistributedCacheManager manager = new TrackerDistributedCacheManager(
+        conf, taskController);
+    JobID jobId = new JobID("jobtracker", 1);
+    manager.newTaskDistributedCacheManager(jobId, conf);
+
+    TaskDistributedCacheManager taskDistributedCacheManager = manager
+        .getTaskDistributedCacheManager(jobId);
+    assertNotNull(taskDistributedCacheManager);
+
+    manager.removeTaskDistributedCacheManager(jobId);
+
+    taskDistributedCacheManager = manager.getTaskDistributedCacheManager(jobId);
+    assertNull(taskDistributedCacheManager);
+  }
+
 }