You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by ch...@apache.org on 2016/09/01 02:33:44 UTC
hbase git commit: HBASE-15278 AsyncRPCClient hangs if Connection
closes before RPC call response
Repository: hbase
Updated Branches:
refs/heads/master 847955427 -> 389255048
HBASE-15278 AsyncRPCClient hangs if Connection closes before RPC call response
Project: http://git-wip-us.apache.org/repos/asf/hbase/repo
Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/38925504
Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/38925504
Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/38925504
Branch: refs/heads/master
Commit: 3892550484c110e53236a6b5c337ce5ef96ee194
Parents: 8479554
Author: chenheng <ch...@apache.org>
Authored: Thu Sep 1 10:33:24 2016 +0800
Committer: chenheng <ch...@apache.org>
Committed: Thu Sep 1 10:33:24 2016 +0800
----------------------------------------------------------------------
.../org/apache/hadoop/hbase/ipc/RpcServer.java | 2 +-
.../hadoop/hbase/ipc/AbstractTestIPC.java | 55 ++++++++++++++++++++
2 files changed, 56 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hbase/blob/38925504/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/RpcServer.java
----------------------------------------------------------------------
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 759da82..c787c98 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
@@ -2677,7 +2677,7 @@ public class RpcServer implements RpcServerInterface, ConfigurationObserver {
}
Connection register(SocketChannel channel) {
- Connection connection = new Connection(channel, System.currentTimeMillis());
+ Connection connection = getConnection(channel, System.currentTimeMillis());
add(connection);
if (LOG.isDebugEnabled()) {
LOG.debug("Server connection from " + connection +
http://git-wip-us.apache.org/repos/asf/hbase/blob/38925504/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/AbstractTestIPC.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/AbstractTestIPC.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/AbstractTestIPC.java
index 771ef93..2e26e13 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/AbstractTestIPC.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/AbstractTestIPC.java
@@ -35,6 +35,8 @@ import com.google.protobuf.ServiceException;
import java.io.IOException;
import java.net.InetSocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.List;
@@ -287,4 +289,57 @@ public abstract class AbstractTestIPC {
rpcServer.stop();
}
}
+
+ static class TestFailingRpcServer extends TestRpcServer {
+
+ TestFailingRpcServer() throws IOException {
+ this(new FifoRpcScheduler(CONF, 1), CONF);
+ }
+
+ TestFailingRpcServer(Configuration conf) throws IOException {
+ this(new FifoRpcScheduler(conf, 1), conf);
+ }
+
+ TestFailingRpcServer(RpcScheduler scheduler, Configuration conf) throws IOException {
+ super(scheduler, conf);
+ }
+
+ class FailingConnection extends Connection {
+ public FailingConnection(SocketChannel channel, long lastContact) {
+ super(channel, lastContact);
+ }
+
+ @Override
+ protected void processRequest(ByteBuffer buf) throws IOException, InterruptedException {
+ // this will throw exception after the connection header is read, and an RPC is sent
+ // from client
+ throw new DoNotRetryIOException("Failing for test");
+ }
+ }
+
+ @Override
+ protected Connection getConnection(SocketChannel channel, long time) {
+ return new FailingConnection(channel, time);
+ }
+ }
+
+ /** Tests that the connection closing is handled by the client with outstanding RPC calls */
+ @Test (timeout = 30000)
+ public void testConnectionCloseWithOutstandingRPCs() throws InterruptedException, IOException {
+ Configuration conf = new Configuration(CONF);
+ RpcServer rpcServer = new TestFailingRpcServer(conf);
+ try (AbstractRpcClient client = createRpcClient(conf)) {
+ rpcServer.start();
+ BlockingInterface stub = newBlockingStub(client, rpcServer.getListenerAddress());
+ EchoRequestProto param = EchoRequestProto.newBuilder().setMessage("hello").build();
+ stub.echo(
+ new PayloadCarryingRpcController(CellUtil.createCellScanner(ImmutableList.<Cell> of(CELL))),
+ param);
+ fail("RPC should have failed because connection closed");
+ } catch (ServiceException e) {
+ LOG.info("Caught expected exception: " + e.toString());
+ } finally {
+ rpcServer.stop();
+ }
+ }
}