HDFS-16728. RBF throw IndexOutOfBoundsException with disableNameServices (#4734). Contributed by ZanderXu.

Reviewed-by: He Xiaoqiao <hexiaoqiao@apache.org>
Reviewed-by: Inigo Goiri <inigoiri@apache.org>
Signed-off-by: Ayush Saxena <ayushsaxena@apache.org>

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterClientProtocol.java

@@ -2214,7 +2214,7 @@ public class RouterClientProtocol implements ClientProtocol {
           .invokeConcurrent(locations, method, false, -1,
               DirectoryListing.class);
       return listings;
-    } catch (RouterResolveException e) {
+    } catch (NoLocationException | RouterResolveException e) {
       LOG.debug("Cannot get locations for {}, {}.", src, e.getMessage());
       return new ArrayList<>();
     }

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterRpcServer.java

@@ -1764,6 +1764,9 @@ public class RouterRpcServer extends AbstractService implements ClientProtocol,
           locs.add(loc);
         }
       }
+      if (locs.isEmpty()) {
+        throw new NoLocationException(path, this.subclusterResolver.getClass());
+      }
       return locs;
     } catch (IOException ioe) {
       if (this.rpcMonitor != null) {

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

@@ -59,6 +59,7 @@ public class MockResolver
   private String defaultNamespace = null;
   private boolean disableDefaultNamespace = false;
   private volatile boolean disableRegistration = false;
+  private TreeSet<String> disableNamespaces = new TreeSet<>();
 
   public MockResolver() {
     this.cleanRegistrations();
@@ -300,9 +301,17 @@ public class MockResolver
     return Collections.unmodifiableSet(this.namespaces);
   }
 
+  public void clearDisableNamespaces() {
+    this.disableNamespaces.clear();
+  }
+
+  public void disableNamespace(String nsId) {
+    this.disableNamespaces.add(nsId);
+  }
+
   @Override
   public Set<String> getDisabledNamespaces() throws IOException {
-    return new TreeSet<>();
+    return this.disableNamespaces;
   }
 
   @Override

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

@@ -1548,6 +1548,24 @@ public class TestRouterRpc {
     }
   }
 
+  @Test
+  public void testMkdirWithDisableNameService() throws Exception {
+    MockResolver resolver = (MockResolver)router.getRouter().getSubclusterResolver();
+    String ns0 = cluster.getNameservices().get(0);
+    resolver.addLocation("/mnt", ns0, "/");
+    MockResolver activeNamenodeResolver = (MockResolver)router.getRouter().getNamenodeResolver();
+    activeNamenodeResolver.disableNamespace(ns0);
+
+    try {
+      FsPermission permission = new FsPermission("777");
+      RouterRpcServer rpcServer = router.getRouter().getRpcServer();
+      LambdaTestUtils.intercept(NoLocationException.class,
+          () -> rpcServer.mkdirs("/mnt/folder0/folder1", permission, true));
+    } finally {
+      activeNamenodeResolver.clearDisableNamespaces();
+    }
+  }
+
   @Test
   public void testProxyExceptionMessages() throws IOException {