HDFS-16296. RouterRpcFairnessPolicyController add rejected permits for each nameservice

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/metrics/FederationRPCMBean.java

@@ -120,4 +120,10 @@ public interface FederationRPCMBean {
    * @return Number of operations rejected due to lack of permits.
    */
   long getProxyOpPermitRejected();
+
+  /**
+   * Get the number of operations rejected due to lack of permits of each namespace.
+   * @return Number of operations rejected due to lack of permits of each namespace.
+   */
+  String getProxyOpPermitRejectedPerNs();
 }

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/metrics/FederationRPCMetrics.java

@@ -286,4 +286,9 @@ public void incrProxyOpPermitRejected() {
   public long getProxyOpPermitRejected() {
     return proxyOpPermitRejected.value();
   }
+
+  @Override
+  public String getProxyOpPermitRejectedPerNs() {
+    return rpcServer.getRPCClient().getRejectedPermitsPerNsJSON();
+  }
 }

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

@@ -47,13 +47,15 @@
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.Callable;
 import java.util.concurrent.CancellationException;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
@@ -134,6 +136,7 @@ public class RouterRpcClient {
 
   /** Fairness manager to control handlers assigned per NS. */
   private RouterRpcFairnessPolicyController routerRpcFairnessPolicyController;
+  private Map<String, AtomicLong> rejectedPermitsPerNs = new ConcurrentHashMap<>();
 
   /**
    * Create a router RPC client to manage remote procedure calls to NNs.
@@ -319,6 +322,15 @@ public String getAsyncCallerPoolJson() {
     return JSON.toString(info);
   }
 
+  /**
+   * JSON representation of the rejected permits for each nameservice.
+   *
+   * @return String representation of the rejected permits for each nameservice.
+   */
+  public String getRejectedPermitsPerNsJSON() {
+    return JSON.toString(rejectedPermitsPerNs);
+  }
+
   /**
    * Get ClientProtocol proxy client for a NameNode. Each combination of user +
    * NN must use a unique proxy client. Previously created clients are cached
@@ -1544,6 +1556,7 @@ private void acquirePermit(
       if (rpcMonitor != null) {
         rpcMonitor.getRPCMetrics().incrProxyOpPermitRejected();
       }
+      incrRejectedPermitForNs(nsId);
       LOG.debug("Permit denied for ugi: {} for method: {}",
           ugi, m.getMethodName());
       String msg =
@@ -1576,4 +1589,14 @@ private void releasePermit(
     return (AbstractRouterRpcFairnessPolicyController
           )routerRpcFairnessPolicyController;
   }
+
+  private void incrRejectedPermitForNs(String ns) {
+    rejectedPermitsPerNs.putIfAbsent(ns, new AtomicLong(0));
+    rejectedPermitsPerNs.get(ns).incrementAndGet();
+  }
+
+  public Long getRejectedPermitForNs(String ns) {
+    return rejectedPermitsPerNs.containsKey(ns) ?
+        rejectedPermitsPerNs.get(ns).get() : 0L;
+  }
 }

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

@@ -146,6 +146,7 @@ private void startLoadTest(final boolean isConcurrent, final boolean fairness)
     Configuration conf = new HdfsConfiguration();
     final int numOps = 10;
     final AtomicInteger overloadException = new AtomicInteger();
+    int originalRejectedPermits = getTotalRejectedPermits(routerContext);
 
     for (int i = 0; i < numOps; i++) {
       DFSClient routerClient = null;
@@ -177,6 +178,9 @@ private void startLoadTest(final boolean isConcurrent, final boolean fairness)
       }
       overloadException.get();
     }
+    int latestRejectedPermits = getTotalRejectedPermits(routerContext);
+    assertEquals(latestRejectedPermits - originalRejectedPermits,
+        overloadException.get());
 
     if (fairness) {
       assertTrue(overloadException.get() > 0);
@@ -208,4 +212,15 @@ private void invokeConcurrent(ClientProtocol routerProto, String clientName)
     routerProto.renewLease(clientName);
   }
 
+  private int getTotalRejectedPermits(RouterContext routerContext) {
+    int totalRejectedPermits = 0;
+    for (String ns : cluster.getNameservices()) {
+      totalRejectedPermits += routerContext.getRouter().getRpcServer()
+          .getRPCClient().getRejectedPermitForNs(ns);
+    }
+    totalRejectedPermits += routerContext.getRouter().getRpcServer()
+        .getRPCClient()
+        .getRejectedPermitForNs(RouterRpcFairnessConstants.CONCURRENT_NS);
+    return totalRejectedPermits;
+  }
 }