HBASE-18559 Add histogram to MetricsConnection to track concurrent calls per server

Signed-off-by: Andrew Purtell <apurtell@apache.org>

hbase-client/src/main/java/org/apache/hadoop/hbase/client/MetricsConnection.java

@@ -73,6 +73,7 @@ public class MetricsConnection implements StatisticTrackable {
     private long responseSizeBytes = 0;
     private long startTime = 0;
     private long callTimeMs = 0;
+    private int concurrentCallsPerServer = 0;
 
     public long getRequestSizeBytes() {
       return requestSizeBytes;
@@ -105,6 +106,14 @@ public class MetricsConnection implements StatisticTrackable {
     public void setCallTimeMs(long callTimeMs) {
       this.callTimeMs = callTimeMs;
     }
+
+    public int getConcurrentCallsPerServer() {
+      return concurrentCallsPerServer;
+    }
+
+    public void setConcurrentCallsPerServer(int callsPerServer) {
+      this.concurrentCallsPerServer = callsPerServer;
+    }
   }
 
   @VisibleForTesting
@@ -271,6 +280,7 @@ public class MetricsConnection implements StatisticTrackable {
   @VisibleForTesting protected final Counter metaCacheNumClearRegion;
   @VisibleForTesting protected final Counter hedgedReadOps;
   @VisibleForTesting protected final Counter hedgedReadWin;
+  @VisibleForTesting protected final Histogram concurrentCallsPerServerHist;
 
   // dynamic metrics
 
@@ -327,6 +337,8 @@ public class MetricsConnection implements StatisticTrackable {
     this.putTracker = new CallTracker(this.registry, "Mutate", "Put", scope);
     this.multiTracker = new CallTracker(this.registry, "Multi", scope);
     this.runnerStats = new RunnerStats(this.registry);
+    this.concurrentCallsPerServerHist = registry.histogram(name(MetricsConnection.class, 
+      "concurrentCallsPerServer", scope));
 
     this.reporter = JmxReporter.forRegistry(this.registry).build();
     this.reporter.start();
@@ -422,6 +434,10 @@ public class MetricsConnection implements StatisticTrackable {
 
   /** Report RPC context to metrics system. */
   public void updateRpc(MethodDescriptor method, Message param, CallStats stats) {
+    int callsPerServer = stats.getConcurrentCallsPerServer();
+    if (callsPerServer > 0) {
+      concurrentCallsPerServerHist.update(callsPerServer);
+    }
     // this implementation is tied directly to protobuf implementation details. would be better
     // if we could dispatch based on something static, ie, request Message type.
     if (method.getService() == ClientService.getDescriptor()) {

hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AbstractRpcClient.java

@@ -416,6 +416,7 @@ public abstract class AbstractRpcClient<T extends RpcConnection> implements RpcC
       if (count > maxConcurrentCallsPerServer) {
         throw new ServerTooBusyException(addr, count);
       }
+      cs.setConcurrentCallsPerServer(count);
       T connection = getConnection(remoteId);
       connection.sendRequest(call, hrc);
     } catch (Exception e) {