HDFS-17935. [ARR] Separate router async handlers for active and observer

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

@@ -82,9 +82,15 @@ public class RBFConfigKeys extends CommonConfigurationKeysPublic {
   public static final String DFS_ROUTER_ASYNC_RPC_NS_HANDLER_COUNT_KEY =
           FEDERATION_ROUTER_ASYNC_RPC_PREFIX + "ns.handler.count";
   public static final String DFS_ROUTER_ASYNC_RPC_NS_HANDLER_COUNT_DEFAULT = "";
+  // Example: ns1:count1,ns2:count2,ns3:count3
+  public static final String DFS_ROUTER_ASYNC_RPC_NS_OBSERVER_HANDLER_COUNT_KEY =
+      FEDERATION_ROUTER_ASYNC_RPC_PREFIX + "ns.observer.handler.count";
   public static final String DFS_ROUTER_ASYNC_RPC_HANDLER_COUNT_KEY =
           FEDERATION_ROUTER_ASYNC_RPC_PREFIX + "handler.count";
   public static final int DFS_ROUTER_ASYNC_RPC_HANDLER_COUNT_DEFAULT = 10;
+  public static final String DFS_ROUTER_ASYNC_RPC_OBSERVER_HANDLER_COUNT_KEY =
+      FEDERATION_ROUTER_ASYNC_RPC_PREFIX + "observer.handler.count";
+  public static final int DFS_ROUTER_ASYNC_RPC_OBSERVER_HANDLER_COUNT_DEFAULT = 0;
   public static final String DFS_ROUTER_ASYNC_RPC_QUEUE_SIZE =
       FEDERATION_ROUTER_ASYNC_RPC_PREFIX + "queue.size";
   public static final int DFS_ROUTER_ASYNC_RPC_QUEUE_SIZE_DEFAULT = 1000;

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

@@ -23,8 +23,10 @@
 import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_ASYNC_RPC_ENABLE_KEY;
 import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_ASYNC_RPC_HANDLER_COUNT_DEFAULT;
 import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_ASYNC_RPC_HANDLER_COUNT_KEY;
-import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_ASYNC_RPC_NS_HANDLER_COUNT_DEFAULT;
 import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_ASYNC_RPC_NS_HANDLER_COUNT_KEY;
+import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_ASYNC_RPC_NS_OBSERVER_HANDLER_COUNT_KEY;
+import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_ASYNC_RPC_OBSERVER_HANDLER_COUNT_DEFAULT;
+import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_ASYNC_RPC_OBSERVER_HANDLER_COUNT_KEY;
 import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_ASYNC_RPC_QUEUE_SIZE;
 import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_ASYNC_RPC_QUEUE_SIZE_DEFAULT;
 import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_ASYNC_RPC_RESPONDER_COUNT_KEY;
@@ -299,6 +301,9 @@ public class RouterRpcServer extends AbstractService implements ClientProtocol,
   private final boolean enableAsync;
   private final Map<String, ThreadPoolExecutor> asyncRouterHandlerExecutors =
       new ConcurrentHashMap<>();
+  private boolean useSeparateAsyncRouterOBHandlerExecutors = false;
+  private final Map<String, ThreadPoolExecutor> asyncRouterOBHandlerExecutors =
+      new ConcurrentHashMap<>();
   private ThreadPoolExecutor routerDefaultAsyncHandlerExecutor;
   private ExecutorService routerAsyncResponderExecutor;
 
@@ -509,44 +514,76 @@ public RouterRpcServer(Configuration conf, Router router,
   public void initAsyncThreadPools(Configuration configuration) {
     Set<String> allConfiguredNS = FederationUtil.getAllConfiguredNS(configuration);
     allConfiguredNS.add(CONCURRENT_NS);
-    Map<String, Integer> nsAsyncActiveHandlerCount = parseNsAsyncHandlerCount(configuration);
-    initAsyncHandlerThreadPools(configuration, allConfiguredNS, nsAsyncActiveHandlerCount);
+    Map<String, Integer> nsAsyncActiveHandlerCount = parseNsAsyncHandlerCount(configuration,
+        DFS_ROUTER_ASYNC_RPC_NS_HANDLER_COUNT_KEY);
+    initAsyncHandlerThreadPools(configuration, allConfiguredNS, false,
+        DFS_ROUTER_ASYNC_RPC_HANDLER_COUNT_KEY, DFS_ROUTER_ASYNC_RPC_HANDLER_COUNT_DEFAULT,
+        nsAsyncActiveHandlerCount, asyncRouterHandlerExecutors);
+
+    Map<String, Integer> nsAsyncObserverHandlerCount = parseNsAsyncHandlerCount(configuration,
+        DFS_ROUTER_ASYNC_RPC_NS_OBSERVER_HANDLER_COUNT_KEY);
+    initAsyncHandlerThreadPools(configuration, allConfiguredNS, true,
+        DFS_ROUTER_ASYNC_RPC_OBSERVER_HANDLER_COUNT_KEY,
+        DFS_ROUTER_ASYNC_RPC_OBSERVER_HANDLER_COUNT_DEFAULT, nsAsyncObserverHandlerCount,
+        asyncRouterOBHandlerExecutors);
+
     initAsyncResponderThreadPools(configuration);
   }
 
-  private void initAsyncHandlerThreadPools(Configuration configuration,
-      Set<String> allConfiguredNS, Map<String, Integer> nsAsyncHandlerCount) {
-    LOG.info("Initializing asynchronous handler thread pools");
+  /**
+   * Initializes async handler executors for each configured namespace.
+   *
+   * @param configuration        router configuration
+   * @param allConfiguredNS      set of all configured namespace IDs
+   * @param useObserver         true if these handlers serve observer namenodes
+   * @param handlerCountKey     key for the default handler count
+   * @param handlerCountDefault default handler count
+   * @param nsAsyncHandlerCount per-namespace handler counts
+   * @param executors           executor map to populate
+   */
+  private void initAsyncHandlerThreadPools(Configuration configuration, Set<String> allConfiguredNS,
+      boolean useObserver, String handlerCountKey, int handlerCountDefault,
+      Map<String, Integer> nsAsyncHandlerCount, Map<String, ThreadPoolExecutor> executors) {
+    String namenodeTypeForLogging = useObserver ? "Observer Namenode" : "Active Namenode";
     int asyncQueueSize = configuration.getInt(DFS_ROUTER_ASYNC_RPC_QUEUE_SIZE,
         DFS_ROUTER_ASYNC_RPC_QUEUE_SIZE_DEFAULT);
     if (asyncQueueSize < 1) {
       throw new IllegalArgumentException("Async queue size must be at least 1");
     }
-    int asyncHandlerCountDefault = configuration.getInt(DFS_ROUTER_ASYNC_RPC_HANDLER_COUNT_KEY,
-        DFS_ROUTER_ASYNC_RPC_HANDLER_COUNT_DEFAULT);
+    int asyncHandlerCountDefault = configuration.getInt(handlerCountKey, handlerCountDefault);
+    // Skip separate observer handlers and let them share with active if not configured
+    if (asyncHandlerCountDefault < 1 && nsAsyncHandlerCount.isEmpty() && useObserver) {
+      LOG.info("Async observer handlers are not configured, skipping...");
+      return;
+    }
+
     if (asyncHandlerCountDefault < 1) {
-      throw new  IllegalArgumentException("Async handler count must be at least 1");
+      throw new IllegalArgumentException("Async handler count must be at least 1");
+    }
+    if (useObserver) {
+      useSeparateAsyncRouterOBHandlerExecutors = true;
     }
+    LOG.info("Initializing asynchronous handler thread pools");
     for (String nsId : allConfiguredNS) {
       int dedicatedHandlers = nsAsyncHandlerCount.getOrDefault(nsId, 0);
       if (dedicatedHandlers <= 0) {
         dedicatedHandlers = asyncHandlerCountDefault;
-        LOG.info("Use default async handler count {} for ns {} to init Executors.",
-            asyncHandlerCountDefault, nsId);
+        LOG.info("Use default async handler count {} for ns {} to init {} Executors.",
+            asyncHandlerCountDefault, nsId, namenodeTypeForLogging);
       } else {
-        LOG.info("Dedicated handlers {} for ns {} to init Executors", dedicatedHandlers, nsId);
+        LOG.info("Dedicated handlers {} for ns {} to init {} Executors", dedicatedHandlers, nsId,
+            namenodeTypeForLogging);
       }
 
       int finalDedicatedHandlers = dedicatedHandlers;
-      asyncRouterHandlerExecutors.computeIfAbsent(nsId,
-          id -> {
-            LinkedBlockingQueue<Runnable> queue = new LinkedBlockingQueue<>(asyncQueueSize);
-            return new ThreadPoolExecutor(finalDedicatedHandlers, finalDedicatedHandlers,
-                0L, TimeUnit.MILLISECONDS, queue,
-                new AsyncThreadFactory("Router Async Handler for " + nsId + " #"));
-          });
-      LOG.info("Assigned {} async handlers with queue size {} to nsId {}", dedicatedHandlers,
-          asyncQueueSize, nsId);
+      executors.computeIfAbsent(nsId, id -> {
+        LinkedBlockingQueue<Runnable> queue = new LinkedBlockingQueue<>(asyncQueueSize);
+        return new ThreadPoolExecutor(finalDedicatedHandlers, finalDedicatedHandlers, 0L,
+            TimeUnit.MILLISECONDS, queue, new AsyncThreadFactory(
+            "Router Async " + namenodeTypeForLogging + " Handler for " + nsId + " #"));
+      });
+      LOG.info("Assigned {} async handlers with queue size {} to nsId {} for {}", dedicatedHandlers,
+          asyncQueueSize, nsId, namenodeTypeForLogging);
     }
 
     if (routerDefaultAsyncHandlerExecutor == null) {
@@ -571,23 +608,27 @@ private void initAsyncResponderThreadPools(Configuration configuration) {
     AsyncRpcProtocolPBUtil.setAsyncResponderExecutor(routerAsyncResponderExecutor);
   }
 
-  private Map<String, Integer> parseNsAsyncHandlerCount(Configuration config) {
-    String configNsHandler = config.get(DFS_ROUTER_ASYNC_RPC_NS_HANDLER_COUNT_KEY,
-        DFS_ROUTER_ASYNC_RPC_NS_HANDLER_COUNT_DEFAULT);
+  /**
+   * Parses per-namespace async handler counts from an ns:count list.
+   * @param config router configuration
+   * @param key configuration key to read
+   * @return map of namespace id -> handler count
+   */
+  private Map<String, Integer> parseNsAsyncHandlerCount(Configuration config, String key) {
+    String configNsHandler =
+        config.get(key, RBFConfigKeys.DFS_ROUTER_ASYNC_RPC_NS_HANDLER_COUNT_DEFAULT);
     Map<String, Integer> nsAsyncHandlerCount = new HashMap<>();
     if (StringUtils.isEmpty(configNsHandler)) {
       LOG.info("No per-namespace async handler counts configured ({}). "
-              + "Will use default handler count for all namespaces.",
-          DFS_ROUTER_ASYNC_RPC_NS_HANDLER_COUNT_KEY);
+          + "Will use default handler count for all namespaces.", key);
       return nsAsyncHandlerCount;
     }
     String[] nsHandlers = configNsHandler.split(",");
     for (String nsHandlerInfo : nsHandlers) {
       String[] nsHandlerItems = nsHandlerInfo.split(":");
-      if (nsHandlerItems.length != 2 || StringUtils.isBlank(nsHandlerItems[0]) ||
-          !StringUtils.isNumeric(nsHandlerItems[1])) {
-        LOG.error("The config key: {} is incorrect! The value is {}.",
-            DFS_ROUTER_ASYNC_RPC_NS_HANDLER_COUNT_KEY, nsHandlerInfo);
+      if (nsHandlerItems.length != 2 || StringUtils.isBlank(nsHandlerItems[0])
+          || !StringUtils.isNumeric(nsHandlerItems[1])) {
+        LOG.error("The config key: {} is incorrect! The value is {}.", key, nsHandlerInfo);
         continue;
       }
       nsAsyncHandlerCount.put(nsHandlerItems[0], Integer.parseInt(nsHandlerItems[1]));
@@ -601,17 +642,27 @@ private Map<String, Integer> parseNsAsyncHandlerCount(Configuration config) {
    * Requires async RPC pools to be initialized (async RPC enabled).
    *
    * @param nsId the namespace identifier
+   * @param useObserver if an observer namenode is used
    * @return the corresponding thread pool
    */
-  public ThreadPoolExecutor getAsyncExecutorForNamespace(String nsId) {
-    ThreadPoolExecutor executor =
-        asyncRouterHandlerExecutors.getOrDefault(nsId, routerDefaultAsyncHandlerExecutor);
+  public ThreadPoolExecutor getAsyncExecutorForNamespace(String nsId, boolean useObserver) {
+    ThreadPoolExecutor executor = getAsyncExecutorForNamespaceInternal(nsId, useObserver);
     if (rpcMonitor != null && executor != null) {
       rpcMonitor.recordAsyncHandlerQueueSize(nsId, executor.getQueue().size());
     }
     return executor;
   }
 
+  private ThreadPoolExecutor getAsyncExecutorForNamespaceInternal(String nsId, boolean useObserver) {
+    if (useObserver && useSeparateAsyncRouterOBHandlerExecutors) {
+      ThreadPoolExecutor observerExecutor = asyncRouterOBHandlerExecutors.get(nsId);
+      if (observerExecutor != null) {
+        return observerExecutor;
+      }
+    }
+    return asyncRouterHandlerExecutors.getOrDefault(nsId, routerDefaultAsyncHandlerExecutor);
+  }
+
   /**
    * Clear expired namespace in the shared RouterStateIdContext.
    */
@@ -725,6 +776,10 @@ protected void serviceStop() throws Exception {
       executor.shutdownNow();
     }
     asyncRouterHandlerExecutors.clear();
+    for (ThreadPoolExecutor executor : asyncRouterOBHandlerExecutors.values()) {
+      executor.shutdownNow();
+    }
+    asyncRouterOBHandlerExecutors.clear();
     if (routerDefaultAsyncHandlerExecutor != null) {
       routerDefaultAsyncHandlerExecutor.shutdownNow();
     }

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

@@ -177,7 +177,7 @@ public Object invokeMethod(
     // transfer threadLocalContext to worker threads of executor.
     ThreadLocalContext threadLocalContext = new ThreadLocalContext();
     asyncComplete(null);
-    // Returns a CompletableFuture with RejectedExecutionException if nsExecutor is full.
+    // Returns a CompletableFuture with RejectedExecutionException if the executor is full.
     asyncApplyUseExecutor((AsyncApplyFunction<Object, Object>) o -> {
       if (LOG.isDebugEnabled()) {
         LOG.debug("Async invoke method : {}, {}, {}, {}", method.getName(), useObserver, namenodes,
@@ -192,7 +192,7 @@ public Object invokeMethod(
         releasePermit(nsid, ugi, method, controller);
         return object;
       });
-    }, router.getRpcServer().getAsyncExecutorForNamespace(nsid));
+    }, router.getRpcServer().getAsyncExecutorForNamespace(nsid, useObserver));
 
     // Catch the RejectedExecutionException and convert it to StandbyException
     asyncCatch((ret, e) -> {

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/resources/hdfs-rbf-default.xml

@@ -66,6 +66,27 @@
     </description>
   </property>
 
+  <property>
+    <name>dfs.federation.router.async.rpc.ns.observer.handler.count</name>
+    <value></value>
+    <description>
+      The number of asynchronous handlers for observer namenodes per nameservice, separated by commas,
+      internally separated by colons. The identifier of nameservice is in dfs.nameservices configuration entry.
+      Such as: ns1:count1,ns2:count2,ns3:count3.
+    </description>
+  </property>
+
+  <property>
+    <name>dfs.federation.router.async.rpc.observer.handler.count</name>
+    <value>0</value>
+    <description>
+      The number of async handlers for the router to handle RPC client requests by observer namenodes.
+      Disabled when set to 0 or negative, and dfs.federation.router.async.rpc.ns.observer.handler.count
+      is empty. When disabled, requests to observer namenodes will use the same handlers defined in
+      dfs.federation.router.async.rpc.handler.count.
+    </description>
+  </property>
+
   <property>
     <name>dfs.federation.router.async.rpc.queue.size</name>
     <value>1000</value>
@@ -150,7 +171,7 @@
     <name>dfs.federation.router.async.rpc.handler.count</name>
     <value>10</value>
     <description>
-      The number of async handler for the router to handle RPC client requests.
+      The number of async handlers for the router to handle RPC client requests by namenodes.
     </description>
   </property>
 

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

@@ -53,6 +53,7 @@
 import static org.apache.hadoop.hdfs.server.federation.metrics.NameserviceRPCMetrics.NAMESERVICE_RPC_METRICS_PREFIX;
 import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_ASYNC_RPC_HANDLER_COUNT_KEY;
 import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_ASYNC_RPC_MAX_ASYNCCALL_PERMIT_KEY;
+import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_ASYNC_RPC_OBSERVER_HANDLER_COUNT_KEY;
 import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_ASYNC_RPC_QUEUE_SIZE;
 import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_ASYNC_RPC_RESPONDER_COUNT_KEY;
 import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_FAIRNESS_ACQUIRE_TIMEOUT;
@@ -95,6 +96,7 @@ public static void setUpCluster() throws Exception {
 
     routerConf.setInt(DFS_ROUTER_ASYNC_RPC_QUEUE_SIZE, QUEUE_CAP);
     routerConf.setInt(DFS_ROUTER_ASYNC_RPC_HANDLER_COUNT_KEY, 1);
+    routerConf.setInt(DFS_ROUTER_ASYNC_RPC_OBSERVER_HANDLER_COUNT_KEY, 1);
     routerConf.setInt(DFS_ROUTER_ASYNC_RPC_RESPONDER_COUNT_KEY, 1);
     routerConf.set(DFS_ROUTER_MONITOR_NAMENODE,
         cluster.getNameservices().get(0) + "," + cluster.getNameservices().get(1));
@@ -114,20 +116,23 @@ public static void setUpCluster() throws Exception {
     FSNamesystem spyNamesystem = NameNodeAdapterMockitoUtil.spyOnNamesystem(nn0.getNamenode());
     // Mock one slow operation. Any public interface from FSNamesystem will do.
     spyNamesystem.writeLock();
-    Mockito.when(spyNamesystem.getFilesBlockingDecom(anyLong(), anyString()))
-        .thenAnswer(invocationOnMock -> {
-          if (testLatch == null) {
+    try {
+      Mockito.when(spyNamesystem.getFilesBlockingDecom(anyLong(), anyString()))
+          .thenAnswer(invocationOnMock -> {
+            if (testLatch == null) {
+              return null;
+            }
+            String invokePath = invocationOnMock.getArgument(1);
+            if (invokePath.startsWith("/veryBigOperation")) {
+              testLatch.await();
+            } else {
+              return invocationOnMock.callRealMethod();
+            }
             return null;
-          }
-          String invokePath = invocationOnMock.getArgument(1);
-          if (invokePath.startsWith("/veryBigOperation")) {
-            testLatch.await();
-          } else {
-            return invocationOnMock.callRealMethod();
-          }
-          return null;
-        });
-    spyNamesystem.writeUnlock();
+          });
+    } finally {
+      spyNamesystem.writeUnlock();
+    }
 
     testLatch = new CountDownLatch(1);
     MiniRouterDFSCluster.RouterContext router = cluster.getRouterContext(ns0, NAMENODES[0]);
@@ -165,7 +170,7 @@ public void testInvokeMethodQueueOverflow() throws Exception {
           asyncRpcClient.getOrderedNamenodes(ns0, true);
       // Downstream namespace processing this huge request
       asyncRpcClient.invokeMethod(ugi, namenodes, true, protocol, method.getMethod(), params);
-      ThreadPoolExecutor nsExecutor = routerRpcServer.getAsyncExecutorForNamespace(ns0);
+      ThreadPoolExecutor nsExecutor = routerRpcServer.getAsyncExecutorForNamespace(ns0, true);
       // Successfully sent this request downstream, but all subsequent ones will get stuck
       GenericTestUtils.waitFor(() -> nsExecutor.getQueue().isEmpty(), 50, 500);
       GenericTestUtils.waitFor(() -> nsExecutor.getCompletedTaskCount() == 1, 50, 500);
@@ -197,7 +202,7 @@ void verifyQueueSize(int size, ThreadPoolExecutor nsExecutor)
     GenericTestUtils.waitFor(() -> {
       try {
         // Ping getAsyncExecutorForNamespace for metrics recording
-        routerRpcServer.getAsyncExecutorForNamespace(ns0);
+        routerRpcServer.getAsyncExecutorForNamespace(ns0, true);
         assertGauge("AsyncHandlerQueueSize", size,
             getMetrics(NAMESERVICE_RPC_METRICS_PREFIX + ns0));
         return true;

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

@@ -58,6 +58,7 @@
 import static org.apache.hadoop.hdfs.server.federation.FederationTestUtils.NAMENODES;
 import static org.apache.hadoop.hdfs.server.federation.MiniRouterDFSCluster.DEFAULT_HEARTBEAT_INTERVAL_MS;
 import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_ASYNC_RPC_HANDLER_COUNT_KEY;
+import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_ASYNC_RPC_OBSERVER_HANDLER_COUNT_KEY;
 import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_ASYNC_RPC_RESPONDER_COUNT_KEY;
 import static org.apache.hadoop.hdfs.server.federation.router.async.utils.AsyncUtil.syncReturn;
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -111,6 +112,7 @@ public static void setUpCluster() throws Exception {
     // Reduce the number of RPC clients threads to overload the Router easy
     routerConf.setInt(RBFConfigKeys.DFS_ROUTER_CLIENT_THREADS_SIZE, 1);
     routerConf.setInt(DFS_ROUTER_ASYNC_RPC_HANDLER_COUNT_KEY, 1);
+    routerConf.setInt(DFS_ROUTER_ASYNC_RPC_OBSERVER_HANDLER_COUNT_KEY, 1);
     routerConf.setInt(DFS_ROUTER_ASYNC_RPC_RESPONDER_COUNT_KEY, 1);
     // We decrease the DN cache times to make the test faster
     routerConf.setTimeDuration(