HADOOP-17024. ListStatus on ViewFS root should list the linkFallBack root

Author: abhishekdas99

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

@@ -123,6 +123,7 @@ static class INodeDir<T> extends INode<T> {
     private final Map<String, INode<T>> children = new HashMap<>();
     private T internalDirFs =  null; //filesystem of this internal directory
     private boolean isRoot = false;
+    private INodeLink<T> fallbackLink = null;
 
     INodeDir(final String pathToNode, final UserGroupInformation aUgi) {
       super(pathToNode, aUgi);
@@ -149,6 +150,17 @@ boolean isRoot() {
       return isRoot;
     }
 
+    INodeLink<T> getFallbackLink() {
+      return fallbackLink;
+    }
+
+    void addFallbackLink(INodeLink<T> fallbackLink) throws IOException {
+      if (!isRoot) {
+        throw new IOException("fallbackLink can only be set for root");
+      }
+      this.fallbackLink = fallbackLink;
+    }
+
     Map<String, INode<T>> getChildren() {
       return Collections.unmodifiableMap(children);
     }
@@ -580,6 +592,7 @@ protected InodeTree(final Configuration config, final String viewName)
         }
       }
       rootFallbackLink = fallbackLink;
+      getRootDir().addFallbackLink(rootFallbackLink);
     }
 
     if (!gotMountTableEntry) {

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

@@ -1204,6 +1204,7 @@ public FileStatus getFileStatus(Path f) throws IOException {
     public FileStatus[] listStatus(Path f) throws AccessControlException,
         FileNotFoundException, IOException {
       checkPathIsSlash(f);
+      FileStatus[] fallbackStatuses = listStatusForFallbackLink();
       FileStatus[] result = new FileStatus[theInternalDir.getChildren().size()];
       int i = 0;
       for (Entry<String, INode<FileSystem>> iEntry :
@@ -1226,7 +1227,39 @@ public FileStatus[] listStatus(Path f) throws AccessControlException,
                 myUri, null));
         }
       }
-      return result;
+      if (fallbackStatuses.length > 0) {
+        return consolidateFileStatuses(fallbackStatuses, result);
+      } else {
+        return result;
+      }
+    }
+
+    private FileStatus[] consolidateFileStatuses(FileStatus[] fallbackStatuses,
+        FileStatus[] mountPointStatuses) {
+      ArrayList<FileStatus> result = new ArrayList<>();
+      Set<String> pathSet = new HashSet<>();
+      int i = 0;
+      for (FileStatus status : mountPointStatuses) {
+        result.add(status);
+        pathSet.add(status.getPath().getName());
+      }
+      for (FileStatus status : fallbackStatuses) {
+        if (!pathSet.contains(status.getPath().getName())) {
+          result.add(status);
+        }
+      }
+      return result.toArray(new FileStatus[0]);
+    }
+
+    private FileStatus[] listStatusForFallbackLink() throws IOException {
+      if (theInternalDir.isRoot() && theInternalDir.getFallbackLink() != null) {
+        URI fallBackUri = theInternalDir.getFallbackLink().targetDirLinkList[0];
+        ChRootedFileSystem linkedFs = (ChRootedFileSystem)
+            theInternalDir.getFallbackLink().getTargetFileSystem();
+        return linkedFs.getMyFs().listStatus(new Path(fallBackUri.toString()));
+      } else {
+        return new FileStatus[0];
+      }
     }
 
     @Override

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

@@ -26,6 +26,7 @@
 import java.net.URI;
 import java.net.URISyntaxException;
 
+import java.util.HashSet;
 import javax.security.auth.login.LoginException;
 
 import org.apache.hadoop.conf.Configuration;
@@ -261,4 +262,84 @@ public void testConfLinkFallbackWithMountPoint() throws Exception {
           e.getMessage().contains(expectedErrorMsg));
     }
   }
+
+  /**
+   * This tests whether the fallback link gets listed for list operation
+   * of root directory of mount table
+   * @throws Exception
+   */
+  @Test
+  public void testListingWithFallbackLink() throws Exception {
+    Path dir1 = new Path(targetTestRoot, "fallbackDir/dir1");
+    fsTarget.mkdirs(dir1);
+    String clusterName = Constants.CONFIG_VIEWFS_DEFAULT_MOUNT_TABLE;
+    URI viewFsUri = new URI(FsConstants.VIEWFS_SCHEME, clusterName,
+        "/", null, null);
+
+    HashSet<Path> beforeFallback = new HashSet<>();
+    try(FileSystem vfs = FileSystem.get(viewFsUri, conf)) {
+      for (FileStatus stat : vfs.listStatus(new Path(viewFsUri.toString()))) {
+        beforeFallback.add(stat.getPath());
+      }
+    }
+
+    ConfigUtil.addLinkFallback(conf, clusterName,
+        new Path(targetTestRoot, "fallbackDir").toUri());
+
+    try (FileSystem vfs = FileSystem.get(viewFsUri, conf)) {
+      HashSet<Path> afterFallback = new HashSet<>();
+      for (FileStatus stat : vfs.listStatus(new Path(viewFsUri.toString()))) {
+        afterFallback.add(stat.getPath());
+      }
+      afterFallback.removeAll(beforeFallback);
+      assertTrue("Listing didn't include fallback link",
+          afterFallback.size() == 1);
+      Path[] fallbackArray = new Path[afterFallback.size()];
+      assertEquals("Path did not match" ,
+          dir1, afterFallback.toArray(fallbackArray)[0]);
+    }
+  }
+
+  /**
+   * This tests whether fallback directory gets shaded during list operation
+   * of root directory of mount table when the same directory name exists as
+   * mount point as well as in the fallback linked directory
+   * @throws Exception
+   */
+  @Test
+  public void testListingWithFallbackLinkWithSameMountDirectories()
+      throws Exception {
+    // Creating two directories under the fallback directory.
+    // "user" directory already exists as configured mount point.
+    Path dir1 = new Path(targetTestRoot, "fallbackDir/user");
+    Path dir2 = new Path(targetTestRoot, "fallbackDir/user1");
+    fsTarget.mkdirs(dir1);
+    fsTarget.mkdirs(dir2);
+    String clusterName = Constants.CONFIG_VIEWFS_DEFAULT_MOUNT_TABLE;
+    URI viewFsUri = new URI(FsConstants.VIEWFS_SCHEME, clusterName,
+        "/", null, null);
+
+    HashSet<Path> beforeFallback = new HashSet<>();
+    try(FileSystem vfs = FileSystem.get(viewFsUri, conf)) {
+      for (FileStatus stat : vfs.listStatus(new Path(viewFsUri.toString()))) {
+        beforeFallback.add(stat.getPath());
+      }
+    }
+    ConfigUtil.addLinkFallback(conf, clusterName,
+        new Path(targetTestRoot, "fallbackDir").toUri());
+
+    try (FileSystem vfs = FileSystem.get(viewFsUri, conf)) {
+      HashSet<Path> afterFallback = new HashSet<>();
+      for (FileStatus stat : vfs.listStatus(new Path(viewFsUri.toString()))) {
+        afterFallback.add(stat.getPath());
+      }
+      afterFallback.removeAll(beforeFallback);
+      assertTrue("The same directory name in fallback link should be shaded",
+          afterFallback.size() == 1);
+      Path[] fallbackArray = new Path[afterFallback.size()];
+      // Only dir2 should be listed as fallback link
+      assertEquals("Path did not match" ,
+          dir2, afterFallback.toArray(fallbackArray)[0]);
+    }
+  }
 }