HDFS-17089. Close child file systems in ViewFileSystem when cache is …

…disabled.

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/InodeTree.java

@@ -413,6 +413,11 @@ public T getTargetFileSystem() throws IOException {
       }
       return targetFileSystem;
     }
+
+    T getTargetFileSystemForClose() throws IOException {
+      return targetFileSystem;
+    }
+
   }
 
   private void createLink(final String src, final String target,

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/ViewFileSystem.java

@@ -1926,12 +1926,41 @@ enum RenameStrategy {
     SAME_FILESYSTEM_ACROSS_MOUNTPOINT
   }
 
+  private void closeChildFileSystems(FileSystem fs) throws IOException {
+    if (fs != null) {
+      FileSystem[] childFs = fs.getChildFileSystems();
+      for (FileSystem child : childFs) {
+        if (child != null) {
+          String disableCacheName = String.format("fs.%s.impl.disable.cache",
+              child.getUri().getScheme());
+          if (config.getBoolean(disableCacheName, false)) {
+            child.close();
+          }
+        }
+      }
+    }
+  }
+
   @Override
   public void close() throws IOException {
     super.close();
     if (enableInnerCache && cache != null) {
       cache.closeAll();
       cache.clear();
     }
+
+    if (!enableInnerCache) {
+      for (InodeTree.MountPoint<FileSystem> mountPoint :
+          fsState.getMountPoints()) {
+        FileSystem targetFs = mountPoint.target.getTargetFileSystemForClose();
+        closeChildFileSystems(targetFs);
+      }
+
+      if (fsState.isRootInternalDir() &&
+          fsState.getRootFallbackLink() != null) {
+        closeChildFileSystems(
+            fsState.getRootFallbackLink().getTargetFileSystem());
+      }
+    }
   }
 }

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/fs/viewfs/TestViewFileSystemClose.java

@@ -0,0 +1,47 @@
+package org.apache.hadoop.fs.viewfs;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.FsConstants;
+import org.apache.hadoop.fs.Path;
+import org.junit.Test;
+
+import java.io.IOException;
+
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+public class TestViewFileSystemClose {
+
+  /**
+   * Verify that all child file systems of a ViewFileSystem will be shut down
+   * when the cache is disabled.
+   * @throws IOException
+   */
+  @Test
+  public void testFileSystemLeak() throws IOException {
+
+    Configuration conf = new Configuration();
+    conf.set("fs.viewfs.impl", ViewFileSystem.class.getName());
+    conf.setBoolean("fs.viewfs.enable.inner.cache", false);
+    conf.setBoolean("fs.viewfs.impl.disable.cache", true);
+    conf.setBoolean("fs.hdfs.impl.disable.cache", true);
+
+    String rootPath = "hdfs://localhost/tmp";
+    ConfigUtil.addLink(conf, "/data", new Path(rootPath, "data").toUri());
+    ViewFileSystem viewFs =
+        (ViewFileSystem) FileSystem.get(FsConstants.VIEWFS_URI, conf);
+
+    FileSystem[] children = viewFs.getChildFileSystems();
+    viewFs.close();
+    FileSystem.closeAll();
+    for (FileSystem fs : children) {
+      try {
+        fs.create(new Path(rootPath, "neverSuccess"));
+        fail();
+      } catch (IOException ioe) {
+        assertTrue(ioe.getMessage().contains("Filesystem closed"));
+      }
+    }
+  }
+}