HBASE-26727 Fix CallDroppedException reporting (#4088)

Signed-off-by: Andrew Purtell <[email protected]>

Author: bbeaudreault

hbase-hadoop-compat/src/main/java/org/apache/hadoop/hbase/metrics/ExceptionTrackingSource.java

@@ -44,6 +44,7 @@ public interface ExceptionTrackingSource extends BaseSource {
   String EXCEPTIONS_QUOTA_EXCEEDED = "exceptions.quotaExceeded";
   String EXCEPTIONS_RPC_THROTTLING = "exceptions.rpcThrottling";
   String EXCEPTIONS_CALL_DROPPED = "exceptions.callDropped";
+  String EXCEPTIONS_CALL_TIMED_OUT = "exceptions.callTimedOut";
 
   void exception();
 
@@ -62,4 +63,5 @@ public interface ExceptionTrackingSource extends BaseSource {
   void quotaExceededException();
   void rpcThrottlingException();
   void callDroppedException();
+  void callTimedOut();
 }

hbase-hadoop2-compat/src/main/java/org/apache/hadoop/hbase/metrics/ExceptionTrackingSourceImpl.java

@@ -41,6 +41,7 @@ public class ExceptionTrackingSourceImpl extends BaseSourceImpl
   protected MutableFastCounter exceptionsQuotaExceeded;
   protected MutableFastCounter exceptionsRpcThrottling;
   protected MutableFastCounter exceptionsCallDropped;
+  protected MutableFastCounter exceptionsCallTimedOut;
 
   public ExceptionTrackingSourceImpl(String metricsName, String metricsDescription,
                                      String metricsContext, String metricsJmxContext) {
@@ -75,6 +76,8 @@ public void init() {
       .newCounter(EXCEPTIONS_RPC_THROTTLING, EXCEPTIONS_TYPE_DESC, 0L);
     this.exceptionsCallDropped = this.getMetricsRegistry()
       .newCounter(EXCEPTIONS_CALL_DROPPED, EXCEPTIONS_TYPE_DESC, 0L);
+    this.exceptionsCallTimedOut = this.getMetricsRegistry()
+      .newCounter(EXCEPTIONS_CALL_TIMED_OUT, EXCEPTIONS_TYPE_DESC, 0L);
   }
 
   @Override
@@ -141,4 +144,9 @@ public void rpcThrottlingException() {
   public void callDroppedException() {
     exceptionsCallDropped.incr();
   }
+
+  @Override
+  public void callTimedOut() {
+    exceptionsCallTimedOut.incr();
+  }
 }

hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/CallRunner.java

@@ -99,6 +99,7 @@ public void run() {
       call.setStartTime(EnvironmentEdgeManager.currentTime());
       if (call.getStartTime() > call.getDeadline()) {
         RpcServer.LOG.warn("Dropping timed out call: " + call);
+        this.rpcServer.getMetrics().callTimedOut();
         return;
       }
       this.status.setStatus("Setting up call");
@@ -207,6 +208,7 @@ public void drop() {
       call.setResponse(null, null, CALL_DROPPED_EXCEPTION, "Call dropped, server "
         + (address != null ? address : "(channel closed)") + " is overloaded, please retry.");
       call.sendResponseIfReady();
+      this.rpcServer.getMetrics().exception(CALL_DROPPED_EXCEPTION);
     } catch (ClosedChannelException cce) {
       InetSocketAddress address = rpcServer.getListenerAddress();
       RpcServer.LOG.warn(Thread.currentThread().getName() + ": caught a ClosedChannelException, " +

hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/MetricsHBaseServer.java

@@ -135,6 +135,10 @@ public void exception(Throwable throwable) {
     }
   }
 
+  void callTimedOut() {
+    source.callTimedOut();
+  }
+
   public MetricsHBaseServerSource getMetricsSource() {
     return source;
   }

hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/TestCallRunner.java

@@ -17,6 +17,8 @@
  */
 package org.apache.hadoop.hbase.ipc;
 
+import java.net.InetSocketAddress;
+import org.apache.hadoop.hbase.CallDroppedException;
 import org.apache.hadoop.hbase.HBaseClassTestRule;
 import org.apache.hadoop.hbase.monitoring.MonitoredRPCHandlerImpl;
 import org.apache.hadoop.hbase.testclassification.RPCTests;
@@ -60,7 +62,7 @@ public void testCallCleanup() {
   }
 
   @Test
-  public void testCallRunnerDrop() {
+  public void testCallRunnerDropDisconnected() {
     RpcServerInterface mockRpcServer = Mockito.mock(RpcServerInterface.class);
     Mockito.when(mockRpcServer.isStarted()).thenReturn(true);
     ServerCall mockCall = Mockito.mock(ServerCall.class);
@@ -71,4 +73,21 @@ public void testCallRunnerDrop() {
     cr.drop();
     Mockito.verify(mockCall, Mockito.times(1)).cleanup();
   }
+
+  @Test
+  public void testCallRunnerDropConnected() {
+    RpcServerInterface mockRpcServer = Mockito.mock(RpcServerInterface.class);
+    MetricsHBaseServer mockMetrics = Mockito.mock(MetricsHBaseServer.class);
+    Mockito.when(mockRpcServer.getMetrics()).thenReturn(mockMetrics);
+    Mockito.when(mockRpcServer.isStarted()).thenReturn(true);
+    Mockito.when(mockRpcServer.getListenerAddress()).thenReturn(InetSocketAddress.createUnresolved("foo", 60020));
+    ServerCall mockCall = Mockito.mock(ServerCall.class);
+    Mockito.when(mockCall.disconnectSince()).thenReturn(-1L);
+
+    CallRunner cr = new CallRunner(mockRpcServer, mockCall);
+    cr.setStatus(new MonitoredRPCHandlerImpl());
+    cr.drop();
+    Mockito.verify(mockCall, Mockito.times(1)).cleanup();
+    Mockito.verify(mockMetrics).exception(Mockito.any(CallDroppedException.class));
+  }
 }