HADOOP-17024. ListStatus on ViewFS root (ls "/") should list the linkFallBack root (configured target root). Contributed by Abhishek Das.

(cherry picked from commit ce4ec74453)
(cherry picked from commit 5b248de42d)

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

@@ -123,6 +123,7 @@ abstract class InodeTree<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 @@ abstract class InodeTree<T> {
       return isRoot;
     }
 
+    INodeLink<T> getFallbackLink() {
+      return fallbackLink;
+    }
+
+    void addFallbackLink(INodeLink<T> link) throws IOException {
+      if (!isRoot) {
+        throw new IOException("Fallback link can only be added for root");
+      }
+      this.fallbackLink = link;
+    }
+
     Map<String, INode<T>> getChildren() {
       return Collections.unmodifiableMap(children);
     }
@@ -580,6 +592,7 @@ abstract class InodeTree<T> {
         }
       }
       rootFallbackLink = fallbackLink;
+      getRootDir().addFallbackLink(rootFallbackLink);
     }
 
     if (!gotMountTableEntry) {

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

@@ -1161,10 +1161,19 @@ public class ViewFileSystem extends FileSystem {
     }
     
 
+    /**
+     * {@inheritDoc}
+     *
+     * Note: listStatus on root("/") considers listing from fallbackLink if
+     * available. If the same directory name is present in configured mount
+     * path as well as in fallback link, then only the configured mount path
+     * will be listed in the returned result.
+     */
     @Override
     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 :
@@ -1187,8 +1196,46 @@ public class ViewFileSystem extends FileSystem {
                 myUri, null));
         }
       }
+      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<>();
+      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) {
+        FileSystem linkedFs =
+            theInternalDir.getFallbackLink().getTargetFileSystem();
+        // Fallback link is only applicable for root
+        FileStatus[] statuses = linkedFs.listStatus(new Path("/"));
+        for (FileStatus status : statuses) {
+          // Fix the path back to viewfs scheme
+          status.setPath(
+              new Path(myUri.toString(), status.getPath().getName()));
+        }
+        return statuses;
+      } else {
+        return new FileStatus[0];
+      }
+    }
 
     @Override
     public boolean mkdirs(Path dir, FsPermission permission)

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

@@ -25,10 +25,12 @@ import java.net.URI;
 import java.net.URISyntaxException;
 import java.util.ArrayList;
 import java.util.EnumSet;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 
+import java.util.Set;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.conf.Configuration;
@@ -943,10 +945,19 @@ public class ViewFs extends AbstractFileSystem {
       return -1;
     }
 
+    /**
+     * {@inheritDoc}
+     *
+     * Note: listStatus on root("/") considers listing from fallbackLink if
+     * available. If the same directory name is present in configured mount
+     * path as well as in fallback link, then only the configured mount path
+     * will be listed in the returned result.
+     */
     @Override
     public FileStatus[] listStatus(final Path f) throws AccessControlException,
         IOException {
       checkPathIsSlash(f);
+      FileStatus[] fallbackStatuses = listStatusForFallbackLink();
       FileStatus[] result = new FileStatus[theInternalDir.getChildren().size()];
       int i = 0;
       for (Entry<String, INode<AbstractFileSystem>> iEntry :
@@ -972,8 +983,46 @@ public class ViewFs extends AbstractFileSystem {
                 myUri, null));
         }
       }
+      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<>();
+      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) {
+        AbstractFileSystem linkedFs =
+            theInternalDir.getFallbackLink().getTargetFileSystem();
+        // Fallback link is only applicable for root
+        FileStatus[] statuses = linkedFs.listStatus(new Path("/"));
+        for (FileStatus status : statuses) {
+          // Fix the path back to viewfs scheme
+          status.setPath(
+              new Path(myUri.toString(), status.getPath().getName()));
+        }
+        return statuses;
+      } else {
+        return new FileStatus[0];
+      }
+    }
 
     @Override
     public void mkdir(final Path dir, final FsPermission permission,

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

@@ -26,6 +26,7 @@ import java.io.IOException;
 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,101 @@ public class TestViewFileSystemLinkFallback extends ViewFileSystemBaseTest {
           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()];
+      afterFallback.toArray(fallbackArray);
+      Path expected = new Path(viewFsUri.toString(), "dir1");
+      assertEquals("Path did not match",
+          expected, fallbackArray[0]);
+
+      // Create a directory using the returned fallback path and verify
+      Path childDir = new Path(fallbackArray[0], "child");
+      vfs.mkdirs(childDir);
+      FileStatus status = fsTarget.getFileStatus(new Path(dir1, "child"));
+      assertTrue(status.isDirectory());
+      assertTrue(vfs.getFileStatus(childDir).isDirectory());
+    }
+  }
+
+  /**
+   * 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 user1 should be listed as fallback link
+      Path expected = new Path(viewFsUri.toString(), "user1");
+      assertEquals("Path did not match",
+          expected, afterFallback.toArray(fallbackArray)[0]);
+
+      // Create a directory using the returned fallback path and verify
+      Path childDir = new Path(fallbackArray[0], "child");
+      vfs.mkdirs(childDir);
+      FileStatus status = fsTarget.getFileStatus(new Path(dir2, "child"));
+      assertTrue(status.isDirectory());
+      assertTrue(vfs.getFileStatus(childDir).isDirectory());
+    }
+  }
 }