diff --git a/hbase-hadoop-compat/src/main/java/org/apache/hadoop/hbase/metrics/ExceptionTrackingSource.java b/hbase-hadoop-compat/src/main/java/org/apache/hadoop/hbase/metrics/ExceptionTrackingSource.java index 2407a733701..68a9c536440 100644 --- a/hbase-hadoop-compat/src/main/java/org/apache/hadoop/hbase/metrics/ExceptionTrackingSource.java +++ b/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(); } diff --git a/hbase-hadoop-compat/src/main/java/org/apache/hadoop/hbase/metrics/ExceptionTrackingSourceImpl.java b/hbase-hadoop-compat/src/main/java/org/apache/hadoop/hbase/metrics/ExceptionTrackingSourceImpl.java index d6dabadd8e8..65214624da5 100644 --- a/hbase-hadoop-compat/src/main/java/org/apache/hadoop/hbase/metrics/ExceptionTrackingSourceImpl.java +++ b/hbase-hadoop-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 class ExceptionTrackingSourceImpl extends BaseSourceImpl .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 class ExceptionTrackingSourceImpl extends BaseSourceImpl public void callDroppedException() { exceptionsCallDropped.incr(); } + + @Override + public void callTimedOut() { + exceptionsCallTimedOut.incr(); + } } 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 9607a71f6be..400b358497f 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 @@ -99,6 +99,7 @@ public class CallRunner { 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 class CallRunner { 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/main/java/org/apache/hadoop/hbase/ipc/MetricsHBaseServer.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/MetricsHBaseServer.java index d79c8eb4f66..5ff14dd790a 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/MetricsHBaseServer.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/MetricsHBaseServer.java @@ -135,6 +135,10 @@ public class MetricsHBaseServer { } } + void callTimedOut() { + source.callTimedOut(); + } + public MetricsHBaseServerSource getMetricsSource() { return source; } 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 75997afde09..49b7caa29ba 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 class TestCallRunner { } @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 class TestCallRunner { 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)); + } }