From de904ffb5ca4f58b4b30ec133b80cc5a88979ae8 Mon Sep 17 00:00:00 2001 From: chenheng Date: Fri, 23 Oct 2015 01:24:24 +0800 Subject: [PATCH] HBASE-14674 Rpc handler / task monitoring seems to be broken after 0.98 --- .../java/org/apache/hadoop/hbase/ipc/CallRunner.java | 18 ++++-------------- .../org/apache/hadoop/hbase/ipc/FifoRpcScheduler.java | 2 ++ .../java/org/apache/hadoop/hbase/ipc/RpcExecutor.java | 16 ++++++++++++++++ .../org/apache/hadoop/hbase/ipc/TestCallRunner.java | 2 ++ .../hadoop/hbase/ipc/TestSimpleRpcScheduler.java | 4 ++++ 5 files changed, 28 insertions(+), 14 deletions(-) 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 38b7c91..c7b2056 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 @@ -58,20 +58,22 @@ public class CallRunner { this.rpcServer = rpcServer; // Add size of the call to queue size. this.rpcServer.addCallSize(call.getSize()); - this.status = getStatus(); } public Call getCall() { return call; } + public void setStatus(MonitoredRPCHandler status) { + this.status = status; + } + /** * Cleanup after ourselves... let go of references. */ private void cleanup() { this.call = null; this.rpcServer = null; - this.status = null; } public void run() { @@ -157,16 +159,4 @@ public class CallRunner { cleanup(); } } - - MonitoredRPCHandler getStatus() { - // It is ugly the way we park status up in RpcServer. Let it be for now. TODO. - MonitoredRPCHandler status = RpcServer.MONITORED_RPC.get(); - if (status != null) { - return status; - } - status = TaskMonitor.get().createRPCStatus(Thread.currentThread().getName()); - status.pause("Waiting for a call"); - RpcServer.MONITORED_RPC.set(status); - return status; - } } diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/FifoRpcScheduler.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/FifoRpcScheduler.java index 8140c1c..d30eae5 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/FifoRpcScheduler.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/FifoRpcScheduler.java @@ -25,6 +25,7 @@ import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import org.apache.hadoop.hbase.ipc.CallRunner; +import org.apache.hadoop.hbase.monitoring.TaskMonitor; /** * A very simple {@code }RpcScheduler} that serves incoming requests in order. @@ -70,6 +71,7 @@ public class FifoRpcScheduler extends RpcScheduler { executor.submit(new Runnable() { @Override public void run() { + task.setStatus(TaskMonitor.get().createRPCStatus(Thread.currentThread().getName())); task.run(); } }); diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/RpcExecutor.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/RpcExecutor.java index 709429d..84ee631 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/RpcExecutor.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/RpcExecutor.java @@ -31,6 +31,8 @@ import org.apache.hadoop.hbase.Abortable; import org.apache.hadoop.hbase.HConstants; import org.apache.hadoop.hbase.classification.InterfaceAudience; import org.apache.hadoop.hbase.classification.InterfaceStability; +import org.apache.hadoop.hbase.monitoring.MonitoredRPCHandler; +import org.apache.hadoop.hbase.monitoring.TaskMonitor; import org.apache.hadoop.util.StringUtils; import com.google.common.base.Preconditions; @@ -116,6 +118,18 @@ public abstract class RpcExecutor { } } + MonitoredRPCHandler getStatus() { + // It is ugly the way we park status up in RpcServer. Let it be for now. TODO. + MonitoredRPCHandler status = RpcServer.MONITORED_RPC.get(); + if (status != null) { + return status; + } + status = TaskMonitor.get().createRPCStatus(Thread.currentThread().getName()); + status.pause("Waiting for a call"); + RpcServer.MONITORED_RPC.set(status); + return status; + } + protected void consumerLoop(final BlockingQueue myQueue) { boolean interrupted = false; double handlerFailureThreshhold = @@ -124,7 +138,9 @@ public abstract class RpcExecutor { try { while (running) { try { + MonitoredRPCHandler status = getStatus(); CallRunner task = myQueue.take(); + task.setStatus(status); try { activeHandlerCount.incrementAndGet(); task.run(); 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 ec06ff3..47c15ae 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,7 @@ */ package org.apache.hadoop.hbase.ipc; +import org.apache.hadoop.hbase.monitoring.MonitoredRPCHandlerImpl; import org.apache.hadoop.hbase.testclassification.RPCTests; import org.apache.hadoop.hbase.testclassification.SmallTests; import org.junit.Test; @@ -35,6 +36,7 @@ public class TestCallRunner { RpcServer.Call mockCall = Mockito.mock(RpcServer.Call.class); mockCall.connection = Mockito.mock(RpcServer.Connection.class); CallRunner cr = new CallRunner(mockRpcServer, mockCall); + cr.setStatus(new MonitoredRPCHandlerImpl()); cr.run(); } } \ No newline at end of file diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/TestSimpleRpcScheduler.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/TestSimpleRpcScheduler.java index 2b7ffb2..087429a 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/TestSimpleRpcScheduler.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/TestSimpleRpcScheduler.java @@ -28,6 +28,7 @@ import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.HConstants; import org.apache.hadoop.hbase.client.Put; +import org.apache.hadoop.hbase.monitoring.MonitoredRPCHandlerImpl; import org.apache.hadoop.hbase.security.User; import org.apache.hadoop.hbase.testclassification.RPCTests; import org.apache.hadoop.hbase.testclassification.SmallTests; @@ -87,6 +88,7 @@ public class TestSimpleRpcScheduler { scheduler.init(CONTEXT); scheduler.start(); CallRunner task = createMockTask(); + task.setStatus(new MonitoredRPCHandlerImpl()); scheduler.dispatch(task); verify(task, timeout(1000)).run(); scheduler.stop(); @@ -121,6 +123,7 @@ public class TestSimpleRpcScheduler { } }; for (CallRunner task : tasks) { + task.setStatus(new MonitoredRPCHandlerImpl()); doAnswer(answerToRun).when(task).run(); } @@ -303,6 +306,7 @@ public class TestSimpleRpcScheduler { private void doAnswerTaskExecution(final CallRunner callTask, final ArrayList results, final int value, final int sleepInterval) { + callTask.setStatus(new MonitoredRPCHandlerImpl()); doAnswer(new Answer() { @Override public Object answer(InvocationOnMock invocation) { -- 2.3.8 (Apple Git-58)