HDFS-16671. RBF: RouterRpcFairnessPolicyController supports configurable permit acquire timeout

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/fairness/AbstractRouterRpcFairnessPolicyController.java

@@ -29,6 +29,9 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_FAIRNESS_ACQUIRE_TIMEOUT_DEFAULT;
+import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_FAIRNESS_ACQUIRE_TIMEOUT_KEY;
+
 /**
  * Base fairness policy that implements @RouterRpcFairnessPolicyController.
  * Internally a map of nameservice to Semaphore is used to control permits.
@@ -42,15 +45,22 @@ public class AbstractRouterRpcFairnessPolicyController
   /** Hash table to hold semaphore for each configured name service. */
   private Map<String, Semaphore> permits;
 
+  private long acquireTimeout = DFS_ROUTER_FAIRNESS_ACQUIRE_TIMEOUT_DEFAULT;
+
   public void init(Configuration conf) {
     this.permits = new HashMap<>();
+    long timeout = conf.getLong(DFS_ROUTER_FAIRNESS_ACQUIRE_TIMEOUT_KEY,
+        DFS_ROUTER_FAIRNESS_ACQUIRE_TIMEOUT_DEFAULT);
+    if (timeout >= 0) {
+      acquireTimeout = timeout;
+    }
   }
 
   @Override
   public boolean acquirePermit(String nsId) {
     try {
       LOG.debug("Taking lock for nameservice {}", nsId);
-      return this.permits.get(nsId).tryAcquire(1, TimeUnit.SECONDS);
+      return this.permits.get(nsId).tryAcquire(acquireTimeout, TimeUnit.MILLISECONDS);
     } catch (InterruptedException e) {
       LOG.debug("Cannot get a permit for nameservice {}", nsId);
     }

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/fairness/StaticRouterRpcFairnessPolicyController.java

@@ -86,7 +86,7 @@ public void init(Configuration conf)
     // Assign remaining handlers equally to remaining name services and
     // general pool if applicable.
     if (!unassignedNS.isEmpty()) {
-      LOG.info("Unassigned ns {}", unassignedNS.toString());
+      LOG.info("Unassigned ns {}", unassignedNS);
       int handlersPerNS = handlerCount / unassignedNS.size();
       LOG.info("Handlers available per ns {}", handlersPerNS);
       for (String nsId : unassignedNS) {
@@ -109,16 +109,15 @@ public void init(Configuration conf)
   }
 
   private static void logAssignment(String nsId, int count) {
-    LOG.info("Assigned {} handlers to nsId {} ",
-        count, nsId);
+    LOG.info("Assigned {} handlers to nsId {} ", count, nsId);
   }
 
-  private void validateHandlersCount(Configuration conf, int handlerCount,
-                                     Set<String> allConfiguredNS) {
+  private void validateHandlersCount(Configuration conf,
+      int handlerCount, Set<String> allConfiguredNS) {
     int totalDedicatedHandlers = 0;
     for (String nsId : allConfiguredNS) {
       int dedicatedHandlers =
-              conf.getInt(DFS_ROUTER_FAIR_HANDLER_COUNT_KEY_PREFIX + nsId, 0);
+          conf.getInt(DFS_ROUTER_FAIR_HANDLER_COUNT_KEY_PREFIX + nsId, 0);
       if (dedicatedHandlers > 0) {
         // Total handlers should not be less than sum of dedicated handlers.
         totalDedicatedHandlers += dedicatedHandlers;

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

@@ -354,6 +354,9 @@ public class RBFConfigKeys extends CommonConfigurationKeysPublic {
       NoRouterRpcFairnessPolicyController.class;
   public static final String DFS_ROUTER_FAIR_HANDLER_COUNT_KEY_PREFIX =
       FEDERATION_ROUTER_FAIRNESS_PREFIX + "handler.count.";
+  public static final String DFS_ROUTER_FAIRNESS_ACQUIRE_TIMEOUT_KEY =
+      FEDERATION_ROUTER_FAIRNESS_PREFIX + "acquire.timeout";
+  public static final long   DFS_ROUTER_FAIRNESS_ACQUIRE_TIMEOUT_DEFAULT = 1000;
 
   // HDFS Router Federation Rename.
   public static final String DFS_ROUTER_FEDERATION_RENAME_PREFIX =

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

@@ -723,6 +723,14 @@
     </description>
   </property>
 
+  <property>
+    <name>dfs.federation.router.fairness.acquire.timeout</name>
+    <value>1000</value>
+    <description>
+      The timeout value of acquiring permit for RouterRpcFairnessPolicyController.
+    </description>
+  </property>
+
   <property>
     <name>dfs.federation.router.federation.rename.bandwidth</name>
     <value>10</value>

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

@@ -23,10 +23,12 @@
 import org.apache.hadoop.hdfs.server.federation.router.FederationUtil;
 import org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys;
 import org.apache.hadoop.test.GenericTestUtils;
+import org.apache.hadoop.util.Time;
 import org.junit.Test;
 import org.slf4j.LoggerFactory;
 
 import static org.apache.hadoop.hdfs.server.federation.fairness.RouterRpcFairnessConstants.CONCURRENT_NS;
+import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_FAIRNESS_ACQUIRE_TIMEOUT_KEY;
 import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_HANDLER_COUNT_KEY;
 import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_MONITOR_NAMENODE;
 import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.DFS_ROUTER_FAIR_HANDLER_COUNT_KEY_PREFIX;
@@ -83,6 +85,29 @@ public void testHandlerAllocationPreconfigured() {
     assertFalse(routerRpcFairnessPolicyController.acquirePermit(CONCURRENT_NS));
   }
 
+  @Test
+  public void testAcquireTimeout() {
+    Configuration conf = createConf(40);
+    conf.setInt(DFS_ROUTER_FAIR_HANDLER_COUNT_KEY_PREFIX + "ns1", 30);
+    conf.setLong(DFS_ROUTER_FAIRNESS_ACQUIRE_TIMEOUT_KEY, 100);
+    RouterRpcFairnessPolicyController routerRpcFairnessPolicyController =
+        FederationUtil.newFairnessPolicyController(conf);
+
+    // ns1 should have 30 permits allocated
+    for (int i = 0; i < 30; i++) {
+      assertTrue(routerRpcFairnessPolicyController.acquirePermit("ns1"));
+    }
+    long acquireBeginTime = Time.monotonicNow();
+    assertFalse(routerRpcFairnessPolicyController.acquirePermit("ns1"));
+    long acquireEndTime = Time.monotonicNow();
+
+    long acquireTime = acquireEndTime - acquireBeginTime;
+
+    // There are some other operations, so acquireTime should more than 100ms.
+    assertTrue(acquireTime > 100);
+    assertTrue(acquireTime < 100 + 50);
+  }
+
   @Test
   public void testAllocationErrorWithZeroHandlers() {
     Configuration conf = createConf(0);