HDFS-17731. [ARR] Add unit test for async RouterAdminServer. (#7409). Contributed by hfutatzhanghb.

Reviewed-by: Jian Zhang <keepromise@apache.org>
Signed-off-by: He Xiaoqiao <hexiaoqiao@apache.org>

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

@@ -409,7 +409,7 @@ public class Router extends CompositeService implements
    * @return New Router RPC Server.
    * @throws IOException If the router RPC server was not started.
    */
-  protected RouterRpcServer createRpcServer() throws IOException {
+  public RouterRpcServer createRpcServer() throws IOException {
     return new RouterRpcServer(this.conf, this, this.getNamenodeResolver(),
         this.getSubclusterResolver());
   }

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

@@ -74,20 +74,21 @@ import org.mockito.Mockito;
  * The administrator interface of the {@link Router} implemented by
  * {@link RouterAdminServer}.
  */
+@SuppressWarnings("checkstyle:visibilitymodifier")
 public class TestRouterAdmin {
 
-  private static StateStoreDFSCluster cluster;
-  private static RouterContext routerContext;
+  protected static StateStoreDFSCluster cluster;
+  protected static RouterContext routerContext;
   public static final String RPC_BEAN =
       "Hadoop:service=Router,name=FederationRPC";
-  private static List<MountTable> mockMountTable;
-  private static StateStoreService stateStore;
-  private static RouterRpcClient mockRpcClient;
+  protected static List<MountTable> mockMountTable;
+  protected static StateStoreService stateStore;
+  protected static RouterRpcClient mockRpcClient;
 
   @BeforeClass
   public static void globalSetUp() throws Exception {
     cluster = new StateStoreDFSCluster(false, 1);
-    // Build and start a router with State Store + admin + RPC
+    // Build and start a router with State Store + admin + RPC.
     Configuration conf = new RouterConfigBuilder()
         .stateStore()
         .admin()
@@ -101,7 +102,7 @@ public class TestRouterAdmin {
     Router router = routerContext.getRouter();
     stateStore = router.getStateStore();
 
-    // Add two name services for testing disabling
+    // Add two name services for testing disabling.
     ActiveNamenodeResolver membership = router.getNamenodeResolver();
     membership.registerNamenode(
         createNamenodeReport("ns0", "nn1", HAServiceState.ACTIVE));
@@ -129,12 +130,12 @@ public class TestRouterAdmin {
       throws IOException, NoSuchFieldException, IllegalAccessException {
     RouterRpcServer spyRpcServer =
         Mockito.spy(routerContext.getRouter().createRpcServer());
-    //Used reflection to set the 'rpcServer field'
+    // Used reflection to set the 'rpcServer field'.
     setField(routerContext.getRouter(), "rpcServer", spyRpcServer);
     Mockito.doReturn(null).when(spyRpcServer).getFileInfo(Mockito.anyString());
 
-    // mock rpc client for destination check when editing mount tables.
-    //spy RPC client and used reflection to set the 'rpcClient' field
+    // Mock rpc client for destination check when editing mount tables.
+    // Spy RPC client and used reflection to set the 'rpcClient' field.
     mockRpcClient = Mockito.spy(spyRpcServer.getRPCClient());
     setField(spyRpcServer, "rpcClient", mockRpcClient);
     RemoteLocation remoteLocation0 =
@@ -172,7 +173,7 @@ public class TestRouterAdmin {
   public void testSetup() throws Exception {
     assertTrue(
         synchronizeRecords(stateStore, mockMountTable, MountTable.class));
-    // Avoid running with random users
+    // Avoid running with random users.
     routerContext.resetAdminClient();
   }
 
@@ -185,18 +186,18 @@ public class TestRouterAdmin {
     RouterClient client = routerContext.getAdminClient();
     MountTableManager mountTable = client.getMountTableManager();
 
-    // Existing mount table size
+    // Existing mount table size.
     List<MountTable> records = getMountTableEntries(mountTable);
     assertEquals(records.size(), mockMountTable.size());
 
-    // Add
+    // Add.
     AddMountTableEntryRequest addRequest =
         AddMountTableEntryRequest.newInstance(newEntry);
     AddMountTableEntryResponse addResponse =
         mountTable.addMountTableEntry(addRequest);
     assertTrue(addResponse.getStatus());
 
-    // New mount table size
+    // New mount table size.
     List<MountTable> records2 = getMountTableEntries(mountTable);
     assertEquals(records2.size(), mockMountTable.size() + 1);
   }
@@ -209,22 +210,22 @@ public class TestRouterAdmin {
     RouterClient client = routerContext.getAdminClient();
     MountTableManager mountTable = client.getMountTableManager();
 
-    // Existing mount table size
+    // Existing mount table size.
     List<MountTable> entries1 = getMountTableEntries(mountTable);
     assertEquals(entries1.size(), mockMountTable.size());
 
-    // Add
+    // Add.
     AddMountTableEntryRequest addRequest =
         AddMountTableEntryRequest.newInstance(newEntry);
     AddMountTableEntryResponse addResponse =
         mountTable.addMountTableEntry(addRequest);
     assertTrue(addResponse.getStatus());
 
-    // New mount table size
+    // New mount table size.
     List<MountTable> entries2 = getMountTableEntries(mountTable);
     assertEquals(entries2.size(), mockMountTable.size() + 1);
 
-    // Add again, should fail
+    // Add again, should fail.
     AddMountTableEntryResponse addResponse2 =
         mountTable.addMountTableEntry(addRequest);
     assertFalse(addResponse2.getStatus());
@@ -240,27 +241,27 @@ public class TestRouterAdmin {
     RouterClient client = routerContext.getAdminClient();
     MountTableManager mountTable = client.getMountTableManager();
 
-    // Existing mount table size
+    // Existing mount table size.
     List<MountTable> records = getMountTableEntries(mountTable);
     assertEquals(records.size(), mockMountTable.size());
 
-    // Add
+    // Add.
     AddMountTableEntryRequest addRequest =
         AddMountTableEntryRequest.newInstance(newEntry);
     AddMountTableEntryResponse addResponse =
         mountTable.addMountTableEntry(addRequest);
     assertTrue(addResponse.getStatus());
 
-    // New mount table size
+    // New mount table size.
     List<MountTable> records2 = getMountTableEntries(mountTable);
     assertEquals(records2.size(), mockMountTable.size() + 1);
 
-    // Check that we have the read only entry
+    // Check that we have the read only entry.
     MountTable record = getMountTableEntry("/readonly");
     assertEquals("/readonly", record.getSourcePath());
     assertTrue(record.isReadOnly());
 
-    // Removing the new entry
+    // Removing the new entry.
     RemoveMountTableEntryRequest removeRequest =
         RemoveMountTableEntryRequest.newInstance("/readonly");
     RemoveMountTableEntryResponse removeResponse =
@@ -287,19 +288,19 @@ public class TestRouterAdmin {
     RouterClient client = routerContext.getAdminClient();
     MountTableManager mountTable = client.getMountTableManager();
 
-    // Add
+    // Add.
     AddMountTableEntryRequest addRequest;
     AddMountTableEntryResponse addResponse;
     addRequest = AddMountTableEntryRequest.newInstance(newEntry);
     addResponse = mountTable.addMountTableEntry(addRequest);
     assertTrue(addResponse.getStatus());
 
-    // Check that we have the read only entry
+    // Check that we have the read only entry.
     MountTable record = getMountTableEntry(mnt);
     assertEquals(mnt, record.getSourcePath());
     assertEquals(order, record.getDestOrder());
 
-    // Removing the new entry
+    // Removing the new entry.
     RemoveMountTableEntryRequest removeRequest =
         RemoveMountTableEntryRequest.newInstance(mnt);
     RemoveMountTableEntryResponse removeResponse =
@@ -313,16 +314,16 @@ public class TestRouterAdmin {
     RouterClient client = routerContext.getAdminClient();
     MountTableManager mountTable = client.getMountTableManager();
 
-    // Existing mount table size
+    // Existing mount table size.
     List<MountTable> entries1 = getMountTableEntries(mountTable);
     assertEquals(entries1.size(), mockMountTable.size());
 
-    // Remove an entry
+    // Remove an entry.
     RemoveMountTableEntryRequest removeRequest =
         RemoveMountTableEntryRequest.newInstance("/");
     mountTable.removeMountTableEntry(removeRequest);
 
-    // New mount table size
+    // New mount table size.
     List<MountTable> entries2 = getMountTableEntries(mountTable);
     assertEquals(entries2.size(), mockMountTable.size() - 1);
   }
@@ -333,7 +334,7 @@ public class TestRouterAdmin {
     RouterClient client = routerContext.getAdminClient();
     MountTableManager mountTable = client.getMountTableManager();
 
-    // Verify starting condition
+    // Verify starting condition.
     MountTable entry = getMountTableEntry("/");
     assertEquals(
         Collections.singletonList(new RemoteLocation("ns0", "/", "/")),
@@ -346,7 +347,7 @@ public class TestRouterAdmin {
         UpdateMountTableEntryRequest.newInstance(updatedEntry);
     mountTable.updateMountTableEntry(updateRequest);
 
-    // Verify edited condition
+    // Verify edited condition.
     entry = getMountTableEntry("/");
     assertEquals(
         Collections.singletonList(new RemoteLocation("ns1", "/", "/")),
@@ -359,11 +360,11 @@ public class TestRouterAdmin {
     RouterClient client = routerContext.getAdminClient();
     MountTableManager mountTable = client.getMountTableManager();
 
-    // Verify size of table
+    // Verify size of table.
     List<MountTable> entries = getMountTableEntries(mountTable);
     assertEquals(mockMountTable.size(), entries.size());
 
-    // Verify all entries are present
+    // Verify all entries are present.
     int matches = 0;
     for (MountTable e : entries) {
       for (MountTable entry : mockMountTable) {
@@ -387,7 +388,7 @@ public class TestRouterAdmin {
 
   @Test
   public void testVerifyFileInDestinations() throws IOException {
-    // this entry has been created in the mock setup
+    // This entry has been created in the mock setup.
     MountTable newEntry = MountTable.newInstance(
         "/testpath", Collections.singletonMap("ns0", "/testdir"),
         Time.now(), Time.now());
@@ -396,7 +397,7 @@ public class TestRouterAdmin {
     List<String> result = adminServer.verifyFileInDestinations(newEntry);
     assertEquals(0, result.size());
 
-    // this entry was not created in the mock
+    // This entry was not created in the mock.
     newEntry = MountTable.newInstance(
         "/testpath", Collections.singletonMap("ns0", "/testdir1"),
         Time.now(), Time.now());
@@ -413,7 +414,7 @@ public class TestRouterAdmin {
    * @throws IOException If the state store could not be accessed.
    */
   private MountTable getMountTableEntry(final String mount) throws IOException {
-    // Refresh the cache
+    // Refresh the cache.
     stateStore.loadCache(MountTableStoreImpl.class, true);
 
     GetMountTableEntriesRequest request =
@@ -422,7 +423,7 @@ public class TestRouterAdmin {
     MountTableManager mountTable = client.getMountTableManager();
     List<MountTable> results = getMountTableEntries(mountTable, request);
     if (results.size() > 0) {
-      // First result is sorted to have the shortest mount string length
+      // First result is sorted to have the shortest mount string length.
       return results.get(0);
     }
     return null;
@@ -449,22 +450,22 @@ public class TestRouterAdmin {
     RouterClient client = routerContext.getAdminClient();
     NameserviceManager nsManager = client.getNameserviceManager();
 
-    // There shouldn't be any name service disabled
+    // There shouldn't be any name service disabled.
     Set<String> disabled = getDisabledNameservices(nsManager);
     assertTrue(disabled.isEmpty());
 
-    // Disable one and see it
+    // Disable one and see it.
     DisableNameserviceRequest disableReq =
         DisableNameserviceRequest.newInstance("ns0");
     DisableNameserviceResponse disableResp =
         nsManager.disableNameservice(disableReq);
     assertTrue(disableResp.getStatus());
-    // Refresh the cache
+    // Refresh the cache.
     disabled = getDisabledNameservices(nsManager);
     assertEquals(1, disabled.size());
     assertTrue(disabled.contains("ns0"));
 
-    // Enable one and we should have no disabled name services
+    // Enable one and we should have no disabled name services.
     EnableNameserviceRequest enableReq =
         EnableNameserviceRequest.newInstance("ns0");
     EnableNameserviceResponse enableResp =
@@ -473,7 +474,7 @@ public class TestRouterAdmin {
     disabled = getDisabledNameservices(nsManager);
     assertTrue(disabled.isEmpty());
 
-    // Non existing name services should fail
+    // Non existing name services should fail.
     disableReq = DisableNameserviceRequest.newInstance("nsunknown");
     disableResp = nsManager.disableNameservice(disableReq);
     assertFalse(disableResp.getStatus());
@@ -503,7 +504,7 @@ public class TestRouterAdmin {
 
   @Test
   public void testNameserviceManagerWithRules() throws Exception{
-    // Try to disable a name service with a kerberos principal name
+    // Try to disable a name service with a kerberos principal name.
     String username = RouterAdminServer.getSuperUser() + "@Example.com";
     DisableNameserviceResponse disableResp =
         testNameserviceManagerUser(username);

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

@@ -0,0 +1,119 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hdfs.server.federation.router.async;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.ha.HAServiceProtocol;
+import org.apache.hadoop.hdfs.protocol.HdfsFileStatus;
+import org.apache.hadoop.hdfs.server.federation.RouterConfigBuilder;
+import org.apache.hadoop.hdfs.server.federation.StateStoreDFSCluster;
+import org.apache.hadoop.hdfs.server.federation.resolver.ActiveNamenodeResolver;
+import org.apache.hadoop.hdfs.server.federation.resolver.RemoteLocation;
+import org.apache.hadoop.hdfs.server.federation.router.RemoteMethod;
+import org.apache.hadoop.hdfs.server.federation.router.Router;
+import org.apache.hadoop.hdfs.server.federation.router.RouterRpcServer;
+import org.apache.hadoop.hdfs.server.federation.router.TestRouterAdmin;
+import org.apache.hadoop.hdfs.server.federation.router.async.utils.AsyncUtil;
+import org.apache.hadoop.util.Lists;
+import org.junit.BeforeClass;
+import org.mockito.Mockito;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.apache.hadoop.hdfs.server.federation.FederationTestUtils.createNamenodeReport;
+import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_ADMIN_MOUNT_CHECK_ENABLE;
+import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_ASYNC_RPC_ENABLE_KEY;
+
+public class TestAsyncRouterAdmin extends TestRouterAdmin {
+
+  @BeforeClass
+  public static void globalSetUp() throws Exception {
+    cluster = new StateStoreDFSCluster(false, 1);
+    // Build and start a router with State Store + admin + RPC.
+    Configuration conf = new RouterConfigBuilder()
+        .stateStore()
+        .admin()
+        .rpc()
+        .build();
+    conf.setBoolean(DFS_ROUTER_ADMIN_MOUNT_CHECK_ENABLE, true);
+    conf.setBoolean(DFS_ROUTER_ASYNC_RPC_ENABLE_KEY, true);
+
+    cluster.addRouterOverrides(conf);
+    cluster.startRouters();
+    routerContext = cluster.getRandomRouter();
+    mockMountTable = cluster.generateMockMountTable();
+    Router router = routerContext.getRouter();
+    stateStore = router.getStateStore();
+
+    // Add two name services for testing disabling.
+    ActiveNamenodeResolver membership = router.getNamenodeResolver();
+    membership.registerNamenode(
+        createNamenodeReport("ns0", "nn1", HAServiceProtocol.HAServiceState.ACTIVE));
+    membership.registerNamenode(
+        createNamenodeReport("ns1", "nn1", HAServiceProtocol.HAServiceState.ACTIVE));
+    stateStore.refreshCaches(true);
+
+    setUpMocks();
+  }
+
+  private static void setUpMocks()
+      throws IOException, NoSuchFieldException, IllegalAccessException {
+    RouterRpcServer spyRpcServer =
+        Mockito.spy(routerContext.getRouter().createRpcServer());
+    // Used reflection to set the 'rpcServer field'.
+    setField(routerContext.getRouter(), "rpcServer", spyRpcServer);
+    Mockito.doReturn(null).when(spyRpcServer).getFileInfo(Mockito.anyString());
+
+    // Mock rpc client for destination check when editing mount tables.
+    // Spy RPC client and used reflection to set the 'rpcClient' field.
+    mockRpcClient = Mockito.spy(spyRpcServer.getRPCClient());
+    setField(spyRpcServer, "rpcClient", mockRpcClient);
+    RemoteLocation remoteLocation0 =
+        new RemoteLocation("ns0", "/testdir", null);
+    RemoteLocation remoteLocation1 =
+        new RemoteLocation("ns1", "/", null);
+    final Map<RemoteLocation, HdfsFileStatus> mockResponse0 = new HashMap<>();
+    final Map<RemoteLocation, HdfsFileStatus> mockResponse1 = new HashMap<>();
+    mockResponse0.put(remoteLocation0,
+        new HdfsFileStatus.Builder().build());
+    Mockito.doAnswer(invocationOnMock -> {
+      AsyncUtil.asyncComplete(mockResponse0);
+      return null;
+    }).when(mockRpcClient).invokeConcurrent(
+        Mockito.eq(Lists.newArrayList(remoteLocation0)),
+        Mockito.any(RemoteMethod.class),
+        Mockito.eq(false),
+        Mockito.eq(false),
+        Mockito.eq(HdfsFileStatus.class)
+    );
+    mockResponse1.put(remoteLocation1,
+        new HdfsFileStatus.Builder().build());
+    Mockito.doAnswer(invocationOnMock -> {
+      AsyncUtil.asyncComplete(mockResponse1);
+      return null;
+    }).when(mockRpcClient).invokeConcurrent(
+        Mockito.eq(Lists.newArrayList(remoteLocation1)),
+        Mockito.any(RemoteMethod.class),
+        Mockito.eq(false),
+        Mockito.eq(false),
+        Mockito.eq(HdfsFileStatus.class)
+    );
+  }
+}