HADOOP-4829. Allow FileSystem shutdown hook to be disabled. Contributed by Todd Lipcon.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/core/trunk@777152 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -122,6 +122,9 @@ Trunk (unreleased changes)
     HADOOP-5643. Adds a way to decommission TaskTrackers while the JobTracker
     is running. (Amar Kamat via ddas)
 
+    HADOOP-4829. Allow FileSystem shutdown hook to be disabled.
+    (Todd Lipcon via tomwhite)
+
   IMPROVEMENTS
 
     HADOOP-4565. Added CombineFileInputFormat to use data locality information

src/core/core-default.xml

@@ -247,6 +247,17 @@
 </property>
 
 
+<property>
+  <name>fs.automatic.close</name>
+  <value>true</value>
+  <description>By default, FileSystem instances are automatically closed at program
+  exit using a JVM shutdown hook. Setting this property to false disables this
+  behavior. This is an advanced option that should only be used by server applications
+  requiring a more carefully orchestrated shutdown sequence.
+  </description>
+</property>
+
+
 <property>
   <name>local.cache.size</name>
   <value>10737418240</value>

src/core/org/apache/hadoop/fs/FileSystem.java

@@ -25,6 +25,7 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.IdentityHashMap;
 import java.util.Iterator;
 import java.util.List;
@@ -70,7 +71,7 @@ public abstract class FileSystem extends Configured implements Closeable {
   public static final Log LOG = LogFactory.getLog(FileSystem.class);
 
   /** FileSystem cache */
-  private static final Cache CACHE = new Cache();
+  static final Cache CACHE = new Cache();
 
   /** The key this instance is stored under in the cache. */
   private Cache.Key key;
@@ -224,17 +225,6 @@ public abstract class FileSystem extends Configured implements Closeable {
     return (LocalFileSystem)newInstance(LocalFileSystem.NAME, conf);
   }
 
-  private static class ClientFinalizer extends Thread {
-    public synchronized void run() {
-      try {
-        FileSystem.closeAll();
-      } catch (IOException e) {
-        LOG.info("FileSystem.closeAll() threw an exception:\n" + e);
-      }
-    }
-  }
-  private static final ClientFinalizer clientFinalizer = new ClientFinalizer();
-
   /**
    * Close all cached filesystems. Be sure those filesystems are not
    * used anymore.
@@ -1409,7 +1399,10 @@ public abstract class FileSystem extends Configured implements Closeable {
 
   /** Caching FileSystem objects */
   static class Cache {
+    private final ClientFinalizer clientFinalizer = new ClientFinalizer();
+
     private final Map<Key, FileSystem> map = new HashMap<Key, FileSystem>();
+    private final Set<Key> toAutoClose = new HashSet<Key>();
 
     /** A variable that makes all objects in the cache unique */
     private static AtomicLong unique = new AtomicLong(1);
@@ -1434,6 +1427,10 @@ public abstract class FileSystem extends Configured implements Closeable {
         }
         fs.key = key;
         map.put(key, fs);
+
+        if (conf.getBoolean("fs.automatic.close", true)) {
+          toAutoClose.add(key);
+        }
       }
       return fs;
     }
@@ -1441,6 +1438,7 @@ public abstract class FileSystem extends Configured implements Closeable {
     synchronized void remove(Key key, FileSystem fs) {
       if (map.containsKey(key) && fs == map.get(key)) {
         map.remove(key);
+        toAutoClose.remove(key);
         if (map.isEmpty() && !clientFinalizer.isAlive()) {
           if (!Runtime.getRuntime().removeShutdownHook(clientFinalizer)) {
             LOG.info("Could not cancel cleanup thread, though no " +
@@ -1451,11 +1449,27 @@ public abstract class FileSystem extends Configured implements Closeable {
     }
 
     synchronized void closeAll() throws IOException {
+      closeAll(false);
+    }
+
+    /**
+     * Close all FileSystem instances in the Cache.
+     * @param onlyAutomatic only close those that are marked for automatic closing
+     */
+    synchronized void closeAll(boolean onlyAutomatic) throws IOException {
       List<IOException> exceptions = new ArrayList<IOException>();
-      for(; !map.isEmpty(); ) {
-        Map.Entry<Key, FileSystem> e = map.entrySet().iterator().next();
-        final Key key = e.getKey();
-        final FileSystem fs = e.getValue();
+
+      // Make a copy of the keys in the map since we'll be modifying
+      // the map while iterating over it, which isn't safe.
+      List<Key> keys = new ArrayList<Key>();
+      keys.addAll(map.keySet());
+
+      for (Key key : keys) {
+        final FileSystem fs = map.get(key);
+
+        if (onlyAutomatic && !toAutoClose.contains(key)) {
+          continue;
+        }
 
         //remove from cache
         remove(key, fs);
@@ -1475,6 +1489,16 @@ public abstract class FileSystem extends Configured implements Closeable {
       }
     }
 
+    private class ClientFinalizer extends Thread {
+      public synchronized void run() {
+        try {
+          closeAll(true);
+        } catch (IOException e) {
+          LOG.info("FileSystem.Cache.closeAll() threw an exception:\n" + e);
+        }
+      }
+    }
+
     /** FileSystem.Cache.Key */
     static class Key {
       final String scheme;

src/test/hdfs-with-mr/org/apache/hadoop/fs/TestFileSystem.java

@@ -22,6 +22,7 @@ import java.io.DataInputStream;
 import java.io.IOException;
 import java.io.OutputStream;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.Random;
 import java.util.List;
 import java.util.ArrayList;
@@ -578,6 +579,37 @@ public class TestFileSystem extends TestCase {
     }
   }
 
+  public void testFsShutdownHook() throws Exception {
+    final Set<FileSystem> closed = Collections.synchronizedSet(new HashSet<FileSystem>());
+    Configuration conf = new Configuration();
+    Configuration confNoAuto = new Configuration();
+
+    conf.setClass("fs.test.impl", TestShutdownFileSystem.class, FileSystem.class);
+    confNoAuto.setClass("fs.test.impl", TestShutdownFileSystem.class, FileSystem.class);
+    confNoAuto.setBoolean("fs.automatic.close", false);
+
+    TestShutdownFileSystem fsWithAuto =
+      (TestShutdownFileSystem)(new Path("test://a/").getFileSystem(conf));
+    TestShutdownFileSystem fsWithoutAuto =
+      (TestShutdownFileSystem)(new Path("test://b/").getFileSystem(confNoAuto));
+
+    fsWithAuto.setClosedSet(closed);
+    fsWithoutAuto.setClosedSet(closed);
+
+    // Different URIs should result in different FS instances
+    assertNotSame(fsWithAuto, fsWithoutAuto);
+
+    FileSystem.CACHE.closeAll(true);
+    assertEquals(1, closed.size());
+    assertTrue(closed.contains(fsWithAuto));
+
+    closed.clear();
+
+    FileSystem.closeAll();
+    assertEquals(1, closed.size());
+    assertTrue(closed.contains(fsWithoutAuto));
+  }
+
 
   public void testCacheKeysAreCaseInsensitive()
     throws Exception
@@ -626,4 +658,18 @@ public class TestFileSystem extends TestCase {
     fs1.close();
     fs2.close();
   }
+
+  public static class TestShutdownFileSystem extends RawLocalFileSystem {
+    private Set<FileSystem> closedSet;
+
+    public void setClosedSet(Set<FileSystem> closedSet) {
+      this.closedSet = closedSet;
+    }
+    public void close() throws IOException {
+      if (closedSet != null) {
+        closedSet.add(this);
+      }
+      super.close();
+    }
+  }
 }