HDFS-15766. RBF: MockResolver.getMountPoints() breaks the semantic of FileSubclusterResolver. Contributed by Jinglun.

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/resolver/FileSubclusterResolver.java

@@ -20,9 +20,14 @@ package org.apache.hadoop.hdfs.server.federation.resolver;
 
 import java.io.IOException;
 import java.util.List;
+import java.util.LinkedList;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.Collection;
 
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.fs.Path;
 
 /**
  * Interface to map a file path in the global name space to a specific
@@ -75,4 +80,51 @@ public interface FileSubclusterResolver {
    * @return Default namespace identifier.
    */
   String getDefaultNamespace();
+
+  /**
+   * Get a list of mount points for a path.
+   *
+   * @param path Path to get the mount points under.
+   * @param mountPoints the mount points to choose.
+   * @return Return empty list if the path is a mount point but there are no
+   *         mount points under the path. Return null if the path is not a mount
+   *         point and there are no mount points under the path.
+   */
+  static List<String> getMountPoints(String path,
+      Collection<String> mountPoints) {
+    Set<String> children = new TreeSet<>();
+    boolean exists = false;
+    for (String subPath : mountPoints) {
+      String child = subPath;
+
+      // Special case for /
+      if (!path.equals(Path.SEPARATOR)) {
+        // Get the children
+        int ini = path.length();
+        child = subPath.substring(ini);
+      }
+
+      if (child.isEmpty()) {
+        // This is a mount point but without children
+        exists = true;
+      } else if (child.startsWith(Path.SEPARATOR)) {
+        // This is a mount point with children
+        exists = true;
+        child = child.substring(1);
+
+        // We only return immediate children
+        int fin = child.indexOf(Path.SEPARATOR);
+        if (fin > -1) {
+          child = child.substring(0, fin);
+        }
+        if (!child.isEmpty()) {
+          children.add(child);
+        }
+      }
+    }
+    if (!exists) {
+      return null;
+    }
+    return new LinkedList<>(children);
+  }
 }

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/resolver/MountTableResolver.java

@@ -452,46 +452,12 @@ public class MountTableResolver
     verifyMountTable();
     final String path = RouterAdmin.normalizeFileSystemPath(str);
 
-    Set<String> children = new TreeSet<>();
     readLock.lock();
     try {
       String from = path;
       String to = path + Character.MAX_VALUE;
       SortedMap<String, MountTable> subMap = this.tree.subMap(from, to);
-
+      return FileSubclusterResolver.getMountPoints(path, subMap.keySet());
-      boolean exists = false;
-      for (String subPath : subMap.keySet()) {
-        String child = subPath;
-
-        // Special case for /
-        if (!path.equals(Path.SEPARATOR)) {
-          // Get the children
-          int ini = path.length();
-          child = subPath.substring(ini);
-        }
-
-        if (child.isEmpty()) {
-          // This is a mount point but without children
-          exists = true;
-        } else if (child.startsWith(Path.SEPARATOR)) {
-          // This is a mount point with children
-          exists = true;
-          child = child.substring(1);
-
-          // We only return immediate children
-          int fin = child.indexOf(Path.SEPARATOR);
-          if (fin > -1) {
-            child = child.substring(0, fin);
-          }
-          if (!child.isEmpty()) {
-            children.add(child);
-          }
-        }
-      }
-      if (!exists) {
-        return null;
-      }
-      return new LinkedList<>(children);
     } finally {
       readLock.unlock();
     }

hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/MockResolver.java

@@ -94,6 +94,16 @@ public class MockResolver
     }
   }
 
+  public boolean removeLocation(String mount, String nsId, String location) {
+    List<RemoteLocation> locationsList = this.locations.get(mount);
+    final RemoteLocation remoteLocation =
+        new RemoteLocation(nsId, location, mount);
+    if (locationsList != null) {
+      return locationsList.remove(remoteLocation);
+    }
+    return false;
+  }
+
   public synchronized void cleanRegistrations() {
     this.resolver = new HashMap<>();
     this.namespaces = new HashSet<>();
@@ -327,33 +337,13 @@ public class MockResolver
 
   @Override
   public List<String> getMountPoints(String path) throws IOException {
-    List<String> mounts = new ArrayList<>();
+    List<String> mountPoints = new ArrayList<>();
-    // for root path search, returning all downstream root level mapping
+    for (String mp : this.locations.keySet()) {
-    if (path.equals("/")) {
+      if (mp.startsWith(path)) {
-      // Mounts only supported under root level
+        mountPoints.add(mp);
-      for (String mount : this.locations.keySet()) {
-        if (mount.length() > 1) {
-          // Remove leading slash, this is the behavior of the mount tree,
-          // return only names.
-          mounts.add(mount.replace("/", ""));
-        }
-      }
-    } else {
-      // a simplified version of MountTableResolver implementation
-      for (String key : this.locations.keySet()) {
-        if (key.startsWith(path)) {
-          String child = key.substring(path.length());
-          if (child.length() > 0) {
-            // only take children so remove parent path and /
-            mounts.add(key.substring(path.length()+1));
-          }
-        }
-      }
-      if (mounts.size() == 0) {
-        mounts = null;
       }
     }
-    return mounts;
+    return FileSubclusterResolver.getMountPoints(path, mountPoints);
   }
 
   @Override

hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterRpc.java

@@ -885,37 +885,40 @@ public class TestRouterRpc {
     resolver.addLocation(mountPoint, ns0, "/");
 
     FsPermission permission = new FsPermission("777");
-    routerProtocol.mkdirs(mountPoint, permission, false);
     routerProtocol.mkdirs(snapshotFolder, permission, false);
-    for (int i = 1; i <= 9; i++) {
+    try {
-      String folderPath = snapshotFolder + "/subfolder" + i;
+      for (int i = 1; i <= 9; i++) {
-      routerProtocol.mkdirs(folderPath, permission, false);
+        String folderPath = snapshotFolder + "/subfolder" + i;
+        routerProtocol.mkdirs(folderPath, permission, false);
+      }
+
+      LOG.info("Create the snapshot: {}", snapshotFolder);
+      routerProtocol.allowSnapshot(snapshotFolder);
+      String snapshotName =
+          routerProtocol.createSnapshot(snapshotFolder, "snap");
+      assertEquals(snapshotFolder + "/.snapshot/snap", snapshotName);
+      assertTrue(
+          verifyFileExists(routerFS, snapshotFolder + "/.snapshot/snap"));
+
+      LOG.info("Rename the snapshot and check it changed");
+      routerProtocol.renameSnapshot(snapshotFolder, "snap", "newsnap");
+      assertFalse(
+          verifyFileExists(routerFS, snapshotFolder + "/.snapshot/snap"));
+      assertTrue(
+          verifyFileExists(routerFS, snapshotFolder + "/.snapshot/newsnap"));
+      LambdaTestUtils.intercept(SnapshotException.class,
+          "Cannot delete snapshot snap from path " + snapshotFolder + ":",
+          () -> routerFS.deleteSnapshot(new Path(snapshotFolder), "snap"));
+
+      LOG.info("Delete the snapshot and check it is not there");
+      routerProtocol.deleteSnapshot(snapshotFolder, "newsnap");
+      assertFalse(
+          verifyFileExists(routerFS, snapshotFolder + "/.snapshot/newsnap"));
+    } finally {
+      // Cleanup
+      assertTrue(routerProtocol.delete(snapshotFolder, true));
+      assertTrue(resolver.removeLocation(mountPoint, ns0, "/"));
     }
-
-    LOG.info("Create the snapshot: {}", snapshotFolder);
-    routerProtocol.allowSnapshot(snapshotFolder);
-    String snapshotName = routerProtocol.createSnapshot(
-        snapshotFolder, "snap");
-    assertEquals(snapshotFolder + "/.snapshot/snap", snapshotName);
-    assertTrue(verifyFileExists(routerFS, snapshotFolder + "/.snapshot/snap"));
-
-    LOG.info("Rename the snapshot and check it changed");
-    routerProtocol.renameSnapshot(snapshotFolder, "snap", "newsnap");
-    assertFalse(
-        verifyFileExists(routerFS, snapshotFolder + "/.snapshot/snap"));
-    assertTrue(
-        verifyFileExists(routerFS, snapshotFolder + "/.snapshot/newsnap"));
-    LambdaTestUtils.intercept(SnapshotException.class,
-        "Cannot delete snapshot snap from path " + snapshotFolder + ":",
-        () -> routerFS.deleteSnapshot(new Path(snapshotFolder), "snap"));
-
-    LOG.info("Delete the snapshot and check it is not there");
-    routerProtocol.deleteSnapshot(snapshotFolder, "newsnap");
-    assertFalse(
-        verifyFileExists(routerFS, snapshotFolder + "/.snapshot/newsnap"));
-
-    // Cleanup
-    routerProtocol.delete(mountPoint, true);
   }
 
   @Test