MAPREDUCE-7094. LocalDistributedCacheManager leaves classloaders open, which leaks FDs. Contributed by Adam Szita.

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapred/LocalDistributedCacheManager.java

@@ -73,6 +73,7 @@ class LocalDistributedCacheManager {
   private List<String> localClasspaths = new ArrayList<String>();
   
   private List<File> symlinksCreated = new ArrayList<File>();
+  private URLClassLoader classLoaderCreated = null;
   
   private boolean setupCalled = false;
   
@@ -82,7 +83,7 @@ class LocalDistributedCacheManager {
    * @param conf
    * @throws IOException
    */
-  public void setup(JobConf conf, JobID jobId) throws IOException {
+  public synchronized void setup(JobConf conf, JobID jobId) throws IOException {
     File workDir = new File(System.getProperty("user.dir"));
     
     // Generate YARN local resources objects corresponding to the distributed
@@ -212,7 +213,7 @@ class LocalDistributedCacheManager {
    * Should be called after setup().
    * 
    */
-  public boolean hasLocalClasspaths() {
+  public synchronized boolean hasLocalClasspaths() {
     if (!setupCalled) {
       throw new IllegalStateException(
           "hasLocalClasspaths() should be called after setup()");
@@ -224,8 +225,11 @@ class LocalDistributedCacheManager {
    * Creates a class loader that includes the designated
    * files and archives.
    */
-  public ClassLoader makeClassLoader(final ClassLoader parent)
+  public synchronized ClassLoader makeClassLoader(final ClassLoader parent)
       throws MalformedURLException {
+    if (classLoaderCreated != null) {
+      throw new IllegalStateException("A classloader was already created");
+    }
     final URL[] urls = new URL[localClasspaths.size()];
     for (int i = 0; i < localClasspaths.size(); ++i) {
       urls[i] = new File(localClasspaths.get(i)).toURI().toURL();
@@ -234,12 +238,29 @@ class LocalDistributedCacheManager {
     return AccessController.doPrivileged(new PrivilegedAction<ClassLoader>() {
       @Override
       public ClassLoader run() {
-        return new URLClassLoader(urls, parent);
+        classLoaderCreated = new URLClassLoader(urls, parent);
+        return classLoaderCreated;
       }
     });
   }
 
-  public void close() throws IOException {
+  public synchronized void close() throws IOException {
+    if(classLoaderCreated != null) {
+      AccessController.doPrivileged(new PrivilegedAction<Void>() {
+        @Override
+        public Void run() {
+          try {
+            classLoaderCreated.close();
+            classLoaderCreated = null;
+          } catch (IOException e) {
+            LOG.warn("Failed to close classloader created " +
+                "by LocalDistributedCacheManager");
+          }
+          return null;
+        }
+      });
+    }
+
     for (File symlink : symlinksCreated) {
       if (!symlink.delete()) {
         LOG.warn("Failed to delete symlink created by the local job runner: " +

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapred/LocalJobRunner.java

@@ -593,10 +593,16 @@ public class LocalJobRunner implements ClientProtocol {
 
       } finally {
         try {
-          fs.delete(systemJobFile.getParent(), true);  // delete submit dir
-          localFs.delete(localJobFile, true);              // delete local copy
-          // Cleanup distributed cache
-          localDistributedCacheManager.close();
+          try {
+            // Cleanup distributed cache
+            localDistributedCacheManager.close();
+          } finally {
+            try {
+              fs.delete(systemJobFile.getParent(), true); // delete submit dir
+            } finally {
+              localFs.delete(localJobFile, true);         // delete local copy
+            }
+          }
         } catch (IOException e) {
           LOG.warn("Error cleaning up "+id+": "+e);
         }