From 13f284b721132a7309800639ad6c7ce81714d8d0 Mon Sep 17 00:00:00 2001
From: Bryan Beaudreault <bbeaudreault@hubspot.com>
Date: Mon, 31 Jan 2022 19:55:11 -0500
Subject: [PATCH] HBASE-26727 Fix CallDroppedException reporting

---
 .../apache/hadoop/hbase/ipc/CallRunner.java   |  1 +
 .../hadoop/hbase/ipc/TestCallRunner.java      | 21 ++++++++++++++++++-
 2 files changed, 21 insertions(+), 1 deletion(-)

diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/CallRunner.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/CallRunner.java
index ef37247f412f..51a42270c01b 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/CallRunner.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/CallRunner.java
@@ -222,6 +222,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, " +
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/TestCallRunner.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/TestCallRunner.java
index 75997afde093..49b7caa29baf 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/TestCallRunner.java
+++ b/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));
+  }
 }