HDFS-17751. [ARR] Add unit tests using asynchronous router rpc for all in org.apache.hadoop.hdfs.server.federation.router.

Author: hfutatzhanghb

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

@@ -2004,10 +2004,11 @@ private static boolean isReadCall(Method method) {
    * Checks and sets last refresh time for a namespace's stateId.
    * Returns true if refresh time is newer than threshold.
    * Otherwise, return false and call should be handled by active namenode.
-   * @param nsId namespaceID
+   * @param nsId namespaceID.
+   * @return true if refresh time is newer than threshold. Otherwise, return false.
    */
   @VisibleForTesting
-  boolean isNamespaceStateIdFresh(String nsId) {
+  public boolean isNamespaceStateIdFresh(String nsId) {
     if (activeNNStateIdRefreshPeriodMs < 0) {
       return true;
     }

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

@@ -64,7 +64,7 @@ public class RouterStateIdContext implements AlignmentContext {
   /** Nameservice specific overrides of the default setting for enabling observer reads. */
   private HashSet<String> observerReadEnabledOverrides = new HashSet<>();
 
-  RouterStateIdContext(Configuration conf) {
+  public RouterStateIdContext(Configuration conf) {
     this.coordinatedMethods = new HashSet<>();
     // For now, only ClientProtocol methods can be coordinated, so only checking
     // against ClientProtocol.

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

@@ -0,0 +1,29 @@
+/**
+ * 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;
+
+public final class TestRouterConstants {
+
+  private TestRouterConstants() {
+
+  }
+
+  public static final String ASYNC_MODE = "ASYNC";
+  public static final String SYNC_MODE = "SYNC";
+}

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

@@ -38,16 +38,24 @@
 import org.apache.hadoop.hdfs.server.federation.store.protocol.RemoveMountTableEntryRequest;
 import org.apache.hadoop.hdfs.server.federation.store.records.MountTable;
 import org.apache.hadoop.test.LambdaTestUtils;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.AfterAll;
-import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.extension.AfterAllCallback;
+import org.junit.jupiter.api.extension.AfterEachCallback;
+import org.junit.jupiter.api.extension.BeforeEachCallback;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.extension.ExtensionContext;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
 
 import java.io.IOException;
+import java.lang.reflect.Method;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 
+import static org.apache.hadoop.hdfs.server.federation.router.TestRouterConstants.ASYNC_MODE;
+import static org.apache.hadoop.hdfs.server.federation.router.TestRouterConstants.SYNC_MODE;
+import static org.apache.hadoop.hdfs.server.federation.router.async.utils.AsyncUtil.syncReturn;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -64,8 +72,23 @@ public class TestRouterMountTableWithoutDefaultNS {
   private static FileSystem nnFs0;
   private static FileSystem nnFs1;
 
-  @BeforeAll
-  public static void globalSetUp() throws Exception {
+  public static StateStoreDFSCluster getCluster() {
+    return cluster;
+  }
+
+  public static void setCluster(StateStoreDFSCluster cluster) {
+    TestRouterMountTableWithoutDefaultNS.cluster = cluster;
+  }
+
+  public static RouterContext getRouterContext() {
+    return routerContext;
+  }
+
+  public static void setRouterContext(RouterContext routerContext) {
+    TestRouterMountTableWithoutDefaultNS.routerContext = routerContext;
+  }
+
+  public static void globalSetUp(String rpcMode) throws Exception {
     // Build and start a federated cluster
     cluster = new StateStoreDFSCluster(false, 2);
     Configuration conf = new RouterConfigBuilder()
@@ -75,6 +98,9 @@ public static void globalSetUp() throws Exception {
         .build();
     conf.setInt(RBFConfigKeys.DFS_ROUTER_ADMIN_MAX_COMPONENT_LENGTH_KEY, 20);
     conf.setBoolean(RBFConfigKeys.DFS_ROUTER_DEFAULT_NAMESERVICE_ENABLE, false);
+    if (rpcMode.equals(ASYNC_MODE)) {
+      conf.setBoolean(RBFConfigKeys.DFS_ROUTER_ASYNC_RPC_ENABLE_KEY, true);
+    }
     cluster.addRouterOverrides(conf);
     cluster.startCluster();
     cluster.startRouters();
@@ -89,17 +115,7 @@ public static void globalSetUp() throws Exception {
     mountTable = (MountTableResolver) router.getSubclusterResolver();
   }
 
-  @AfterAll
-  public static void tearDown() {
-    if (cluster != null) {
-      cluster.stopRouter(routerContext);
-      cluster.shutdown();
-      cluster = null;
-    }
-  }
-
-  @AfterEach
-  public void clearMountTable() throws IOException {
+  public static void clearMountTable() throws IOException {
     RouterClient client = routerContext.getAdminClient();
     MountTableManager mountTableManager = client.getMountTableManager();
     GetMountTableEntriesRequest req1 = GetMountTableEntriesRequest.newInstance("/");
@@ -129,11 +145,90 @@ private boolean addMountTable(final MountTable entry) throws IOException {
     return addResponse.getStatus();
   }
 
+  @Nested
+  @ExtendWith(RouterServerHelperInTestRouterMountTableWithoutDefaultNS.class)
+  class TestWithAsyncRouterRpc {
+    @ParameterizedTest
+    @ValueSource(strings = {ASYNC_MODE})
+    public void testGetFileInfoWithSubMountPointAsync() throws IOException {
+      testGetFileInfoWithSubMountPoint();
+    }
+
+    @ParameterizedTest
+    @ValueSource(strings = {ASYNC_MODE})
+    public void testGetFileInfoWithoutSubMountPointAsync() throws Exception {
+      testGetFileInfoWithoutSubMountPoint(ASYNC_MODE);
+    }
+
+    @ParameterizedTest
+    @ValueSource(strings = {ASYNC_MODE})
+    public void testGetContentSummaryWithSubMountPointAsync() throws Exception {
+      testGetContentSummaryWithSubMountPoint(ASYNC_MODE);
+    }
+
+    @ParameterizedTest
+    @ValueSource(strings = {ASYNC_MODE})
+    public void testGetAllLocationsAsync() throws IOException {
+      testGetAllLocations();
+    }
+
+    @ParameterizedTest
+    @ValueSource(strings = {ASYNC_MODE})
+    public void testGetLocationsForContentSummaryAsync() throws Exception {
+      testGetLocationsForContentSummary();
+    }
+
+    @ParameterizedTest
+    @ValueSource(strings = {ASYNC_MODE})
+    public void testGetContentSummaryAsync() throws Exception {
+      testGetContentSummary(ASYNC_MODE);
+    }
+  }
+
+  @Nested
+  @ExtendWith(RouterServerHelperInTestRouterMountTableWithoutDefaultNS.class)
+  class TestWithSyncRouterRpc {
+    @ParameterizedTest
+    @ValueSource(strings = {SYNC_MODE})
+    public void testGetFileInfoWithSubMountPointSync() throws IOException {
+      testGetFileInfoWithSubMountPoint();
+    }
+
+    @ParameterizedTest
+    @ValueSource(strings = {SYNC_MODE})
+    public void testGetFileInfoWithoutSubMountPointSync() throws Exception {
+      testGetFileInfoWithoutSubMountPoint(SYNC_MODE);
+    }
+
+    @ParameterizedTest
+    @ValueSource(strings = {SYNC_MODE})
+    public void testGetContentSummaryWithSubMountPointSync() throws Exception {
+      testGetContentSummaryWithSubMountPoint(SYNC_MODE);
+    }
+
+    @ParameterizedTest
+    @ValueSource(strings = {SYNC_MODE})
+    public void testGetAllLocationsSync() throws IOException {
+      testGetAllLocations();
+    }
+
+    @ParameterizedTest
+    @ValueSource(strings = {SYNC_MODE})
+    public void testGetLocationsForContentSummarySync() throws Exception {
+      testGetLocationsForContentSummary();
+    }
+
+    @ParameterizedTest
+    @ValueSource(strings = {SYNC_MODE})
+    public void testGetContentSummarySync() throws Exception {
+      testGetContentSummary(SYNC_MODE);
+    }
+  }
+
   /**
    * Verify that RBF that disable default nameservice should support
    * get information about ancestor mount points.
    */
-  @Test
   public void testGetFileInfoWithSubMountPoint() throws IOException {
     MountTable addEntry = MountTable.newInstance("/testdir/1",
         Collections.singletonMap("ns0", "/testdir/1"));
@@ -148,21 +243,26 @@ public void testGetFileInfoWithSubMountPoint() throws IOException {
    * Verify that RBF doesn't support get the file information
    * with no location and sub mount points.
    */
-  @Test
-  public void testGetFileInfoWithoutSubMountPoint() throws Exception {
+  public void testGetFileInfoWithoutSubMountPoint(String rpcMode) throws Exception {
     MountTable addEntry = MountTable.newInstance("/testdir/1",
         Collections.singletonMap("ns0", "/testdir/1"));
     assertTrue(addMountTable(addEntry));
     LambdaTestUtils.intercept(RouterResolveException.class,
-        () -> routerContext.getRouter().getRpcServer().getFileInfo("/testdir2"));
+        () -> {
+          if (rpcMode.equals(ASYNC_MODE)) {
+            routerContext.getRouter().getRpcServer().getFileInfo("/testdir2");
+            syncReturn(HdfsFileStatus.class);
+          } else {
+            routerContext.getRouter().getRpcServer().getFileInfo("/testdir2");
+          }
+        });
   }
 
   /**
    * Verify that RBF that disable default nameservice should support
    * get information about ancestor mount points.
    */
-  @Test
-  public void testGetContentSummaryWithSubMountPoint() throws IOException {
+  public void testGetContentSummaryWithSubMountPoint(String rpcMode) throws Exception {
     MountTable addEntry = MountTable.newInstance("/testdir/1/2",
         Collections.singletonMap("ns0", "/testdir/1/2"));
     assertTrue(addMountTable(addEntry));
@@ -171,7 +271,13 @@ public void testGetContentSummaryWithSubMountPoint() throws IOException {
       writeData(nnFs0, new Path("/testdir/1/2/3"), 10 * 1024 * 1024);
 
       RouterRpcServer routerRpcServer = routerContext.getRouterRpcServer();
-      ContentSummary summaryFromRBF = routerRpcServer.getContentSummary("/testdir");
+      ContentSummary summaryFromRBF;
+      if (rpcMode.equals(ASYNC_MODE)) {
+        routerRpcServer.getContentSummary("/testdir");
+        summaryFromRBF = syncReturn(ContentSummary.class); 
+      } else {
+        summaryFromRBF = routerRpcServer.getContentSummary("/testdir");
+      }
       assertNotNull(summaryFromRBF);
       assertEquals(1, summaryFromRBF.getFileCount());
       assertEquals(10 * 1024 * 1024, summaryFromRBF.getLength());
@@ -180,7 +286,6 @@ public void testGetContentSummaryWithSubMountPoint() throws IOException {
     }
   }
 
-  @Test
   public void testGetAllLocations() throws IOException {
     // Add mount table entry.
     MountTable addEntry = MountTable.newInstance("/testA",
@@ -198,7 +303,6 @@ public void testGetAllLocations() throws IOException {
     assertEquals(3, locations.size());
   }
 
-  @Test
   public void testGetLocationsForContentSummary() throws Exception {
     // Add mount table entry.
     MountTable addEntry = MountTable.newInstance("/testA/testB",
@@ -227,8 +331,7 @@ public void testGetLocationsForContentSummary() throws Exception {
         () -> protocol.getLocationsForContentSummary("/testB"));
   }
 
-  @Test
-  public void testGetContentSummary() throws Exception {
+  public void testGetContentSummary(String rpcMode) throws Exception {
     try {
       // Add mount table entry.
       MountTable addEntry = MountTable.newInstance("/testA",
@@ -246,7 +349,13 @@ public void testGetContentSummary() throws Exception {
       writeData(nnFs1, new Path("/testA/testB/testC/file3"), 1024 * 1024);
 
       RouterRpcServer routerRpcServer = routerContext.getRouterRpcServer();
-      ContentSummary summary = routerRpcServer.getContentSummary("/testA");
+      ContentSummary summary;
+      if (rpcMode.equals(ASYNC_MODE)) {
+        routerRpcServer.getContentSummary("/testA");
+        summary = syncReturn(ContentSummary.class);
+      } else {
+        summary = routerRpcServer.getContentSummary("/testA");
+      }
       assertEquals(3, summary.getFileCount());
       assertEquals(1024 * 1024 * 3, summary.getLength());
 
@@ -266,3 +375,40 @@ void writeData(FileSystem fs, Path path, int fileLength) throws IOException {
     }
   }
 }
+
+class RouterServerHelperInTestRouterMountTableWithoutDefaultNS implements
+    AfterAllCallback, BeforeEachCallback, AfterEachCallback {
+  public static final ThreadLocal<RouterServerHelperInTestRouterMountTableWithoutDefaultNS>
+      TEST_ROUTER_SERVER_TL = new InheritableThreadLocal<>();
+
+  @Override
+  public void afterAll(ExtensionContext context) {
+    if (TestRouterMountTableWithoutDefaultNS.getCluster() != null) {
+      TestRouterMountTableWithoutDefaultNS.getCluster().stopRouter(
+          TestRouterMountTableWithoutDefaultNS.getRouterContext());
+      TestRouterMountTableWithoutDefaultNS.getCluster().shutdown();
+      TestRouterMountTableWithoutDefaultNS.setCluster(null);
+    }
+    TEST_ROUTER_SERVER_TL.remove();
+  }
+
+  @Override
+  public void afterEach(ExtensionContext context) throws IOException {
+    TestRouterMountTableWithoutDefaultNS.clearMountTable();
+  }
+
+  @Override
+  public void beforeEach(ExtensionContext context) throws Exception {
+    Method testMethod = context.getRequiredTestMethod();
+    ValueSource enumAnnotation = testMethod.getAnnotation(ValueSource.class);
+    if (enumAnnotation != null) {
+      String[] strings = enumAnnotation.strings();
+      for (String rpcMode : strings) {
+        if (TEST_ROUTER_SERVER_TL.get() == null) {
+          TestRouterMountTableWithoutDefaultNS.globalSetUp(rpcMode);
+        }
+      }
+    }
+    TEST_ROUTER_SERVER_TL.set(RouterServerHelperInTestRouterMountTableWithoutDefaultNS.this);
+  }
+}

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

@@ -41,6 +41,8 @@
 import java.util.Map;
 
 import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_HTTP_ENABLE;
+import static org.apache.hadoop.hdfs.server.federation.router.TestRouterConstants.ASYNC_MODE;
+import static org.apache.hadoop.hdfs.server.federation.router.TestRouterConstants.SYNC_MODE;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.apache.hadoop.hdfs.server.federation.router.TestRouterNetworkTopologyServlet.clusterNoDatanodes;
@@ -53,16 +55,13 @@ public class TestRouterNetworkTopologyServlet {
   public static StateStoreDFSCluster clusterWithDatanodes;
   public static StateStoreDFSCluster clusterNoDatanodes;
 
-  public static final String ASYNC_MODE = "ASYNC";
-  public static final String SYNC_MODE = "SYNC";
-
   public static void setUp(String rpcMode) throws Exception {
     // Builder configuration.
     Configuration routerConf =
         new RouterConfigBuilder().stateStore().admin().quota().rpc().build();
     routerConf.set(DFS_ROUTER_HTTP_ENABLE, "true");
     // Use async router rpc.
-    if (rpcMode.equals("ASYNC")) {
+    if (rpcMode.equals(ASYNC_MODE)) {
       routerConf.setBoolean(RBFConfigKeys.DFS_ROUTER_ASYNC_RPC_ENABLE_KEY, true);
     }
     Configuration hdfsConf = new Configuration(false);