diff --git hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AsyncRpcChannel.java hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AsyncRpcChannel.java index 787aa47..848548d 100644 --- hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AsyncRpcChannel.java +++ hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AsyncRpcChannel.java @@ -216,6 +216,12 @@ public class AsyncRpcChannel { ch.pipeline() .addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4)); ch.pipeline().addLast(new AsyncServerResponseHandler(this)); + ch.closeFuture().addListener(new GenericFutureListener() { + @Override + public void operationComplete(ChannelFuture future) throws Exception { + close(null); + } + }); try { writeChannelHeader(ch).addListener(new GenericFutureListener() { @Override diff --git hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AsyncServerResponseHandler.java hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AsyncServerResponseHandler.java index 8f6c85b..0073a83 100644 --- hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AsyncServerResponseHandler.java +++ hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AsyncServerResponseHandler.java @@ -121,4 +121,11 @@ public class AsyncServerResponseHandler extends ChannelInboundHandlerAdapter { e.getPort(), doNotRetry) : new RemoteWithExtrasException(innerExceptionClassName, e.getStackTrace(), doNotRetry); } + + @Override + public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { + channel.close(cause); + } + + } diff --git hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/AbstractTestIPC.java hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/AbstractTestIPC.java index ffe4d40..bfbfa8c 100644 --- hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/AbstractTestIPC.java +++ hbase-server/src/test/java/org/apache/hadoop/hbase/ipc/AbstractTestIPC.java @@ -30,6 +30,8 @@ import java.net.ConnectException; import java.net.InetAddress; import java.net.InetSocketAddress; import java.net.SocketTimeoutException; +import java.nio.ByteBuffer; +import java.nio.channels.SocketChannel; import java.util.ArrayList; import java.util.List; @@ -39,6 +41,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.HConstants; import org.apache.hadoop.hbase.KeyValue; @@ -137,13 +140,17 @@ public abstract class AbstractTestIPC { static class TestRpcServer extends RpcServer { TestRpcServer() throws IOException { - this(new FifoRpcScheduler(CONF, 1)); + this(new FifoRpcScheduler(CONF, 1), CONF); + } + + TestRpcServer(Configuration conf) throws IOException { + this(new FifoRpcScheduler(conf, 1), conf); } - TestRpcServer(RpcScheduler scheduler) throws IOException { + TestRpcServer(RpcScheduler scheduler, Configuration conf) throws IOException { super(null, "testRpcServer", Lists .newArrayList(new BlockingServiceAndInterface(SERVICE, null)), new InetSocketAddress( - "localhost", 0), CONF, scheduler); + "localhost", 0), conf, scheduler); } @Override @@ -154,6 +161,39 @@ public abstract class AbstractTestIPC { } } + 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); + } + + } + protected abstract AbstractRpcClient createRpcClientNoCodec(Configuration conf); /** @@ -267,7 +307,7 @@ public abstract class AbstractTestIPC { @Test public void testRpcScheduler() throws IOException, InterruptedException { RpcScheduler scheduler = spy(new FifoRpcScheduler(CONF, 1)); - RpcServer rpcServer = new TestRpcServer(scheduler); + RpcServer rpcServer = new TestRpcServer(scheduler, CONF); verify(scheduler).init((RpcScheduler.Context) anyObject()); AbstractRpcClient client = createRpcClient(CONF); try { @@ -292,6 +332,66 @@ public abstract class AbstractTestIPC { } } + /** Tests that RPC max request size is respected from the server side */ + @Test (timeout = 30000) + public void testRpcMaxRequestSize() throws IOException, InterruptedException { + Configuration conf = new Configuration(CONF); + conf.setInt(RpcServer.MAX_REQUEST_SIZE, 100); + RpcServer rpcServer = new TestRpcServer(conf); + AbstractRpcClient client = createRpcClient(conf); + try { + rpcServer.start(); + MethodDescriptor md = SERVICE.getDescriptorForType().findMethodByName("echo"); + // set total RPC size bigger than 100 bytes + EchoRequestProto param = EchoRequestProto.newBuilder().setMessage("hello.hello.hello.hello." + + "hello.hello.hello.hello.hello.hello.hello.hello.hello.hello.hello.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 it exceeds max request size"); + } catch(IOException ex) { + // pass + } + } finally { + rpcServer.stop(); + } + } + + /** 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 server closed connection"); + } catch(IOException ex) { + // pass + } + } finally { + rpcServer.stop(); + } + } + /** * Instance of RpcServer that echoes client hostAddress back to client */