From d5d81d675ace2d87c4ac19562b6b0a29da3d8902 Mon Sep 17 00:00:00 2001 From: Enis Soztutar Date: Tue, 27 Oct 2015 14:24:21 -0700 Subject: [PATCH] HBASE-14674 Rpc handler / task monitoring seems to be broken after 0.98 (Heng Chen) --- .../apache/hadoop/hbase/ipc/CallRunner.java | 19 ++++--------------- .../hadoop/hbase/ipc/FifoRpcScheduler.java | 2 +- .../apache/hadoop/hbase/ipc/RpcExecutor.java | 3 +++ .../apache/hadoop/hbase/ipc/RpcServer.java | 16 +++++++++++++++- .../hadoop/hbase/ipc/TestCallRunner.java | 2 ++ .../hbase/ipc/TestSimpleRpcScheduler.java | 4 ++++ 6 files changed, 29 insertions(+), 17 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 ede4b4ebcd3..5b52521b93c 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 @@ -25,7 +25,6 @@ import org.apache.hadoop.hbase.CellScanner; import org.apache.hadoop.hbase.classification.InterfaceAudience; import org.apache.hadoop.hbase.ipc.RpcServer.Call; import org.apache.hadoop.hbase.monitoring.MonitoredRPCHandler; -import org.apache.hadoop.hbase.monitoring.TaskMonitor; import org.apache.hadoop.hbase.util.Pair; import org.apache.hadoop.security.UserGroupInformation; import org.apache.hadoop.util.StringUtils; @@ -59,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() { @@ -160,16 +161,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 8140c1c1017..621a8efe7c2 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 @@ -24,7 +24,6 @@ import java.io.IOException; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; -import org.apache.hadoop.hbase.ipc.CallRunner; /** * A very simple {@code }RpcScheduler} that serves incoming requests in order. @@ -70,6 +69,7 @@ public class FifoRpcScheduler extends RpcScheduler { executor.submit(new Runnable() { @Override public void run() { + task.setStatus(RpcServer.getStatus()); 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 709429d720a..27750a7cd47 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,7 @@ 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.util.StringUtils; import com.google.common.base.Preconditions; @@ -124,7 +125,9 @@ public abstract class RpcExecutor { try { while (running) { try { + MonitoredRPCHandler status = RpcServer.getStatus(); CallRunner task = myQueue.take(); + task.setStatus(status); try { activeHandlerCount.incrementAndGet(); task.run(); diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/RpcServer.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/RpcServer.java index c20e972d29a..6cbce9504f8 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/RpcServer.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/RpcServer.java @@ -84,6 +84,7 @@ import org.apache.hadoop.hbase.exceptions.RegionMovedException; import org.apache.hadoop.hbase.io.ByteBufferOutputStream; import org.apache.hadoop.hbase.io.BoundedByteBufferPool; import org.apache.hadoop.hbase.monitoring.MonitoredRPCHandler; +import org.apache.hadoop.hbase.monitoring.TaskMonitor; import org.apache.hadoop.hbase.protobuf.ProtobufUtil; import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.VersionInfo; import org.apache.hadoop.hbase.protobuf.generated.RPCProtos.CellBlockMeta; @@ -2314,7 +2315,8 @@ public class RpcServer implements RpcServerInterface { * @param user client user * @param connection incoming connection * @param addr InetAddress of incoming connection - * @throws org.apache.hadoop.security.authorize.AuthorizationException when the client isn't authorized to talk the protocol + * @throws org.apache.hadoop.security.authorize.AuthorizationException + * when the client isn't authorized to talk the protocol */ public void authorize(UserGroupInformation user, ConnectionHeader connection, InetAddress addr) throws AuthorizationException { @@ -2498,6 +2500,18 @@ public class RpcServer implements RpcServerInterface { return bsasi == null? null: bsasi.getBlockingService(); } + static 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; + } + /** Returns the remote side ip address when invoked inside an RPC * Returns null incase of an error. * @return InetAddress 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 ec06ff3dfdb..47c15ae040b 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 2b7ffb2ad24..087429a744a 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) {