From b03024bc90d46b8e6290da29b59c8eb56b861bfd Mon Sep 17 00:00:00 2001 From: chenheng Date: Wed, 31 Aug 2016 15:45:35 +0800 Subject: [PATCH] HBASE-15278 AsyncRPCClient hangs if Connection closes before RPC call response --- .../org/apache/hadoop/hbase/ipc/RpcServer.java | 2 +- .../apache/hadoop/hbase/ipc/AbstractTestIPC.java | 67 ++++++++++++++++++++++ 2 files changed, 68 insertions(+), 1 deletion(-) 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 + 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 be5ad56..e7ea83a 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 @@ -37,6 +37,8 @@ import com.google.protobuf.ServiceException; import java.io.IOException; import java.net.InetAddress; import java.net.InetSocketAddress; +import java.nio.ByteBuffer; +import java.nio.channels.SocketChannel; import java.util.ArrayList; import java.util.List; @@ -46,6 +48,7 @@ import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.Cell; import org.apache.hadoop.hbase.CellScanner; import org.apache.hadoop.hbase.CellUtil; +import org.apache.hadoop.hbase.DoNotRetryIOException; import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.HBaseTestingUtility; import org.apache.hadoop.hbase.HConstants; @@ -391,4 +394,68 @@ 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 IOException, InterruptedException { + Configuration conf = new Configuration(CONF); + RpcServer rpcServer = new TestFailingRpcServer(conf); + AbstractRpcClient client = createRpcClient(conf); + try { + rpcServer.start(); + MethodDescriptor md = SERVICE.getDescriptorForType().findMethodByName("echo"); + EchoRequestProto param = EchoRequestProto.newBuilder().setMessage("hello").build(); + InetSocketAddress address = rpcServer.getListenerAddress(); + if (address == null) { + throw new IOException("Listener channel is closed"); + } + try { + client.call(new PayloadCarryingRpcController( + CellUtil.createCellScanner(ImmutableList.of(CELL))), md, param, + md.getOutputType().toProto(), User.getCurrent(), address, + new MetricsConnection.CallStats()); + fail("RPC should have failed because connection closed"); + } catch (IOException e) { + // pass + } + } finally { + rpcServer.stop(); + } + } + } -- 2.9.3