From b3660d014708de3d0fb04c9c152934f6020a65ae Mon Sep 17 00:00:00 2001
From: Uma Maheswara Rao G <umamahesh@apache.org>
Date: Thu, 3 Sep 2020 11:06:20 -0700
Subject: [PATCH] HDFS-15529: getChildFilesystems should include fallback fs as
 well (#2234). Contributed by Uma Maheswara Rao G.

---
 .../org/apache/hadoop/fs/viewfs/InodeTree.java   |  9 +++++++++
 .../apache/hadoop/fs/viewfs/ViewFileSystem.java  |  6 ++++++
 ...ewFileSystemOverloadSchemeWithHdfsScheme.java | 16 +++++++++++++---
 3 files changed, 28 insertions(+), 3 deletions(-)

diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/InodeTree.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/InodeTree.java
index 003694f2e9..1f1adea6db 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/InodeTree.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/InodeTree.java
@@ -394,6 +394,15 @@ private boolean hasFallbackLink() {
     return rootFallbackLink != null;
   }
 
+  /**
+   * @return true if the root represented as internalDir. In LinkMergeSlash,
+   * there will be root to root mapping. So, root does not represent as
+   * internalDir.
+   */
+  protected boolean isRootInternalDir() {
+    return root.isInternalDir();
+  }
+
   protected INodeLink<T> getRootFallbackLink() {
     Preconditions.checkState(root.isInternalDir());
     return rootFallbackLink;
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/ViewFileSystem.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/ViewFileSystem.java
index 8c659d1b15..1ba91b5edf 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/ViewFileSystem.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/ViewFileSystem.java
@@ -939,6 +939,12 @@ public FileSystem[] getChildFileSystems() {
       FileSystem targetFs = mountPoint.target.targetFileSystem;
       children.addAll(Arrays.asList(targetFs.getChildFileSystems()));
     }
+
+    if (fsState.isRootInternalDir() && fsState.getRootFallbackLink() != null) {
+      children.addAll(Arrays.asList(
+          fsState.getRootFallbackLink().targetFileSystem
+              .getChildFileSystems()));
+    }
     return children.toArray(new FileSystem[]{});
   }
   
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/fs/viewfs/TestViewFileSystemOverloadSchemeWithHdfsScheme.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/fs/viewfs/TestViewFileSystemOverloadSchemeWithHdfsScheme.java
index 31674f8d13..9a858e17eb 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/fs/viewfs/TestViewFileSystemOverloadSchemeWithHdfsScheme.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/fs/viewfs/TestViewFileSystemOverloadSchemeWithHdfsScheme.java
@@ -476,10 +476,18 @@ public void testViewFsOverloadSchemeWithInnerCache()
     // 2. Two hdfs file systems should be there if no cache.
     conf.setBoolean(Constants.CONFIG_VIEWFS_ENABLE_INNER_CACHE, false);
     try (FileSystem vfs = FileSystem.get(conf)) {
-      Assert.assertEquals(2, vfs.getChildFileSystems().length);
+      Assert.assertEquals(isFallBackExist(conf) ? 3 : 2,
+          vfs.getChildFileSystems().length);
     }
   }
 
+  // HDFS-15529: if any extended tests added fallback, then getChildFileSystems
+  // will include fallback as well.
+  private boolean isFallBackExist(Configuration config) {
+    return config.get(ConfigUtil.getConfigViewFsPrefix(defaultFSURI
+        .getAuthority()) + "." + Constants.CONFIG_VIEWFS_LINK_FALLBACK) != null;
+  }
+
   /**
    * Create mount links as follows
    * hdfs://localhost:xxx/HDFSUser0 --> hdfs://localhost:xxx/HDFSUser/
@@ -501,7 +509,8 @@ public void testViewFsOverloadSchemeWithNoInnerCacheAndHdfsTargets()
     conf.setBoolean(Constants.CONFIG_VIEWFS_ENABLE_INNER_CACHE, false);
     // Two hdfs file systems should be there if no cache.
     try (FileSystem vfs = FileSystem.get(conf)) {
-      Assert.assertEquals(2, vfs.getChildFileSystems().length);
+      Assert.assertEquals(isFallBackExist(conf) ? 3 : 2,
+          vfs.getChildFileSystems().length);
     }
   }
 
@@ -528,7 +537,8 @@ public void testViewFsOverloadSchemeWithNoInnerCacheAndLocalSchemeTargets()
     // cache should work.
     conf.setBoolean(Constants.CONFIG_VIEWFS_ENABLE_INNER_CACHE, false);
     try (FileSystem vfs = FileSystem.get(conf)) {
-      Assert.assertEquals(1, vfs.getChildFileSystems().length);
+      Assert.assertEquals(isFallBackExist(conf) ? 2 : 1,
+          vfs.getChildFileSystems().length);
     }
   }