HBASE-15520 Fix broken TestAsyncIPC
Project: http://git-wip-us.apache.org/repos/asf/hbase/repo Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/fd5c0934 Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/fd5c0934 Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/fd5c0934 Branch: refs/heads/hbase-12439 Commit: fd5c0934b60664ecdde21a994910953339c7060d Parents: 925c185 Author: zhangduo <zhang...@apache.org> Authored: Thu Mar 24 10:24:34 2016 +0800 Committer: zhangduo <zhang...@apache.org> Committed: Thu Mar 24 10:46:44 2016 +0800 ---------------------------------------------------------------------- .../hadoop/hbase/ipc/AsyncRpcChannel.java | 271 +++++++++---------- .../hbase/ipc/AsyncServerResponseHandler.java | 122 +++++---- .../hadoop/hbase/ipc/AbstractTestIPC.java | 3 +- 3 files changed, 188 insertions(+), 208 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hbase/blob/fd5c0934/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AsyncRpcChannel.java ---------------------------------------------------------------------- diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AsyncRpcChannel.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AsyncRpcChannel.java index 787aa47..53eb824 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AsyncRpcChannel.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AsyncRpcChannel.java @@ -17,18 +17,6 @@ */ package org.apache.hadoop.hbase.ipc; -import io.netty.bootstrap.Bootstrap; -import io.netty.buffer.ByteBuf; -import io.netty.buffer.ByteBufOutputStream; -import io.netty.channel.Channel; -import io.netty.channel.ChannelFuture; -import io.netty.channel.ChannelFutureListener; -import io.netty.handler.codec.LengthFieldBasedFrameDecoder; -import io.netty.util.Timeout; -import io.netty.util.TimerTask; -import io.netty.util.concurrent.GenericFutureListener; -import io.netty.util.concurrent.Promise; - import java.io.IOException; import java.net.ConnectException; import java.net.InetSocketAddress; @@ -52,6 +40,7 @@ import org.apache.hadoop.hbase.client.MetricsConnection; import org.apache.hadoop.hbase.exceptions.ConnectionClosingException; import org.apache.hadoop.hbase.protobuf.ProtobufUtil; import org.apache.hadoop.hbase.protobuf.generated.AuthenticationProtos; +import org.apache.hadoop.hbase.protobuf.generated.AuthenticationProtos.TokenIdentifier.Kind; import org.apache.hadoop.hbase.protobuf.generated.RPCProtos; import org.apache.hadoop.hbase.protobuf.generated.TracingProtos; import org.apache.hadoop.hbase.security.AuthMethod; @@ -75,6 +64,18 @@ import com.google.protobuf.Descriptors; import com.google.protobuf.Message; import com.google.protobuf.RpcCallback; +import io.netty.bootstrap.Bootstrap; +import io.netty.buffer.ByteBuf; +import io.netty.buffer.ByteBufOutputStream; +import io.netty.channel.Channel; +import io.netty.channel.ChannelFuture; +import io.netty.channel.ChannelFutureListener; +import io.netty.handler.codec.LengthFieldBasedFrameDecoder; +import io.netty.util.Timeout; +import io.netty.util.TimerTask; +import io.netty.util.concurrent.GenericFutureListener; +import io.netty.util.concurrent.Promise; + /** * Netty RPC channel */ @@ -84,12 +85,12 @@ public class AsyncRpcChannel { private static final int MAX_SASL_RETRIES = 5; - protected final static Map<AuthenticationProtos.TokenIdentifier.Kind, TokenSelector<? extends - TokenIdentifier>> tokenHandlers = new HashMap<>(); + protected final static Map<Kind, TokenSelector<? extends TokenIdentifier>> TOKEN_HANDDLERS + = new HashMap<>(); static { - tokenHandlers.put(AuthenticationProtos.TokenIdentifier.Kind.HBASE_AUTH_TOKEN, - new AuthenticationTokenSelector()); + TOKEN_HANDDLERS.put(AuthenticationProtos.TokenIdentifier.Kind.HBASE_AUTH_TOKEN, + new AuthenticationTokenSelector()); } final AsyncRpcClient client; @@ -111,7 +112,6 @@ public class AsyncRpcChannel { private Token<? extends TokenIdentifier> token; private String serverPrincipal; - // NOTE: closed and connected flags below are only changed when a lock on pendingCalls private final Map<Integer, AsyncCall> pendingCalls = new HashMap<Integer, AsyncCall>(); private boolean connected = false; @@ -128,15 +128,14 @@ public class AsyncRpcChannel { /** * Constructor for netty RPC channel - * * @param bootstrap to construct channel on - * @param client to connect with + * @param client to connect with * @param ticket of user which uses connection * @param serviceName name of service to connect to * @param address to connect to */ - public AsyncRpcChannel(Bootstrap bootstrap, final AsyncRpcClient client, User ticket, String - serviceName, InetSocketAddress address) { + public AsyncRpcChannel(Bootstrap bootstrap, final AsyncRpcClient client, User ticket, + String serviceName, InetSocketAddress address) { this.client = client; this.ticket = ticket; @@ -145,16 +144,12 @@ public class AsyncRpcChannel { this.channel = connect(bootstrap).channel(); - name = ("IPC Client (" + channel.hashCode() + ") to " + - address.toString() + - ((ticket == null) ? - " from unknown user" : - (" from " + ticket.getName()))); + name = ("IPC Client (" + channel.hashCode() + ") to " + address.toString() + + ((ticket == null) ? " from unknown user" : (" from " + ticket.getName()))); } /** * Connect to channel - * * @param bootstrap to connect to * @return future of connection */ @@ -209,12 +204,11 @@ public class AsyncRpcChannel { /** * Start HBase connection - * * @param ch channel to start connection on */ private void startHBaseConnection(Channel ch) { - ch.pipeline() - .addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4)); + ch.pipeline().addLast("frameDecoder", + new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4)); ch.pipeline().addLast(new AsyncServerResponseHandler(this)); try { writeChannelHeader(ch).addListener(new GenericFutureListener<ChannelFuture>() { @@ -248,7 +242,8 @@ public class AsyncRpcChannel { private SaslClientHandler getSaslHandler(final UserGroupInformation realTicket, final Bootstrap bootstrap) throws IOException { return new SaslClientHandler(realTicket, authMethod, token, serverPrincipal, - client.fallbackAllowed, client.conf.get("hbase.rpc.protection", + client.fallbackAllowed, + client.conf.get("hbase.rpc.protection", SaslUtil.QualityOfProtection.AUTHENTICATION.name().toLowerCase()), new SaslClientHandler.SaslExceptionHandler() { @Override @@ -258,7 +253,7 @@ public class AsyncRpcChannel { handleSaslConnectionFailure(retryCount, cause, realTicket); retryOrClose(bootstrap, failureCounter++, random.nextInt(reloginMaxBackoff) + 1, - cause); + cause); } catch (IOException | InterruptedException e) { close(e); } @@ -273,13 +268,12 @@ public class AsyncRpcChannel { /** * Retry to connect or close - * - * @param bootstrap to connect with - * @param failureCount failure count - * @param e exception of fail + * @param bootstrap to connect with + * @param failureCount failure count + * @param e exception of fail */ - private void retryOrClose(final Bootstrap bootstrap, int failureCount, - long timeout, Throwable e) { + private void retryOrClose(final Bootstrap bootstrap, int failureCount, long timeout, + Throwable e) { if (failureCount < client.maxRetries) { client.newTimeout(new TimerTask() { @Override @@ -303,9 +297,8 @@ public class AsyncRpcChannel { public Promise<Message> callMethod(final Descriptors.MethodDescriptor method, final PayloadCarryingRpcController controller, final Message request, final Message responsePrototype, MetricsConnection.CallStats callStats) { - final AsyncCall call = - new AsyncCall(channel.eventLoop(), client.callIdCnt.getAndIncrement(), method, request, - controller, responsePrototype, callStats); + final AsyncCall call = new AsyncCall(channel.eventLoop(), client.callIdCnt.getAndIncrement(), + method, request, controller, responsePrototype, callStats); controller.notifyOnCancel(new RpcCallback<Object>() { @Override public void run(Object parameter) { @@ -331,9 +324,7 @@ public class AsyncRpcChannel { pendingCalls.put(call.id, call); // Add timeout for cleanup if none is present if (cleanupTimer == null && call.getRpcTimeout() > 0) { - cleanupTimer = - client.newTimeout(timeoutTask, call.getRpcTimeout(), - TimeUnit.MILLISECONDS); + cleanupTimer = client.newTimeout(timeoutTask, call.getRpcTimeout(), TimeUnit.MILLISECONDS); } if (!connected) { return call; @@ -351,14 +342,13 @@ public class AsyncRpcChannel { /** * Write the channel header - * * @param channel to write to * @return future of write * @throws java.io.IOException on failure to write */ private ChannelFuture writeChannelHeader(Channel channel) throws IOException { - RPCProtos.ConnectionHeader.Builder headerBuilder = - RPCProtos.ConnectionHeader.newBuilder().setServiceName(serviceName); + RPCProtos.ConnectionHeader.Builder headerBuilder = RPCProtos.ConnectionHeader.newBuilder() + .setServiceName(serviceName); RPCProtos.UserInformation userInfoPB = buildUserInfo(ticket.getUGI(), authMethod); if (userInfoPB != null) { @@ -375,7 +365,6 @@ public class AsyncRpcChannel { headerBuilder.setVersionInfo(ProtobufUtil.getVersionInfo()); RPCProtos.ConnectionHeader header = headerBuilder.build(); - int totalSize = IPCUtil.getTotalSizeWhenWrittenDelimited(header); ByteBuf b = channel.alloc().directBuffer(totalSize); @@ -388,20 +377,19 @@ public class AsyncRpcChannel { /** * Write request to channel - * - * @param call to write + * @param call to write */ private void writeRequest(final AsyncCall call) { try { final RPCProtos.RequestHeader.Builder requestHeaderBuilder = RPCProtos.RequestHeader .newBuilder(); - requestHeaderBuilder.setCallId(call.id) - .setMethodName(call.method.getName()).setRequestParam(call.param != null); + requestHeaderBuilder.setCallId(call.id).setMethodName(call.method.getName()) + .setRequestParam(call.param != null); if (Trace.isTracing()) { Span s = Trace.currentSpan(); - requestHeaderBuilder.setTraceInfo(TracingProtos.RPCTInfo.newBuilder(). - setParentId(s.getSpanId()).setTraceId(s.getTraceId())); + requestHeaderBuilder.setTraceInfo(TracingProtos.RPCTInfo.newBuilder() + .setParentId(s.getSpanId()).setTraceId(s.getTraceId())); } ByteBuffer cellBlock = client.buildCellBlock(call.controller.cellScanner()); @@ -411,7 +399,7 @@ public class AsyncRpcChannel { cellBlockBuilder.setLength(cellBlock.limit()); requestHeaderBuilder.setCellBlockMeta(cellBlockBuilder.build()); } - // Only pass priority if there one. Let zero be same as no priority. + // Only pass priority if there one. Let zero be same as no priority. if (call.controller.getPriority() != PayloadCarryingRpcController.PRIORITY_UNSET) { requestHeaderBuilder.setPriority(call.controller.getPriority()); } @@ -424,7 +412,7 @@ public class AsyncRpcChannel { } ByteBuf b = channel.alloc().directBuffer(4 + totalSize); - try(ByteBufOutputStream out = new ByteBufOutputStream(b)) { + try (ByteBufOutputStream out = new ByteBufOutputStream(b)) { call.callStats.setRequestSizeBytes(IPCUtil.write(out, rh, call.param, cellBlock)); } @@ -436,7 +424,6 @@ public class AsyncRpcChannel { /** * Set up server authorization - * * @throws java.io.IOException if auth setup failed */ private void setupAuthorization() throws IOException { @@ -447,10 +434,10 @@ public class AsyncRpcChannel { if (useSasl && securityInfo != null) { AuthenticationProtos.TokenIdentifier.Kind tokenKind = securityInfo.getTokenKind(); if (tokenKind != null) { - TokenSelector<? extends TokenIdentifier> tokenSelector = tokenHandlers.get(tokenKind); + TokenSelector<? extends TokenIdentifier> tokenSelector = TOKEN_HANDDLERS.get(tokenKind); if (tokenSelector != null) { - token = tokenSelector - .selectToken(new Text(client.clusterId), ticket.getUGI().getTokens()); + token = tokenSelector.selectToken(new Text(client.clusterId), + ticket.getUGI().getTokens()); } else if (LOG.isDebugEnabled()) { LOG.debug("No token selector found for type " + tokenKind); } @@ -460,7 +447,7 @@ public class AsyncRpcChannel { throw new IOException("Can't obtain server Kerberos config key from SecurityInfo"); } this.serverPrincipal = SecurityUtil.getServerPrincipal(client.conf.get(serverKey), - address.getAddress().getCanonicalHostName().toLowerCase()); + address.getAddress().getCanonicalHostName().toLowerCase()); if (LOG.isDebugEnabled()) { LOG.debug("RPC Server Kerberos principal name for service=" + serviceName + " is " + serverPrincipal); @@ -476,16 +463,15 @@ public class AsyncRpcChannel { } if (LOG.isDebugEnabled()) { - LOG.debug("Use " + authMethod + " authentication for service " + serviceName + - ", sasl=" + useSasl); + LOG.debug( + "Use " + authMethod + " authentication for service " + serviceName + ", sasl=" + useSasl); } reloginMaxBackoff = client.conf.getInt("hbase.security.relogin.maxbackoff", 5000); } /** * Build the user information - * - * @param ugi User Group Information + * @param ugi User Group Information * @param authMethod Authorization method * @return UserInformation protobuf */ @@ -499,7 +485,7 @@ public class AsyncRpcChannel { // Send effective user for Kerberos auth userInfoPB.setEffectiveUser(ugi.getUserName()); } else if (authMethod == AuthMethod.SIMPLE) { - //Send both effective user and real user for simple auth + // Send both effective user and real user for simple auth userInfoPB.setEffectiveUser(ugi.getUserName()); if (ugi.getRealUser() != null) { userInfoPB.setRealUser(ugi.getRealUser().getUserName()); @@ -510,8 +496,7 @@ public class AsyncRpcChannel { /** * Create connection preamble - * - * @param byteBuf to write to + * @param byteBuf to write to * @param authMethod to write */ private void createPreamble(ByteBuf byteBuf, AuthMethod authMethod) { @@ -520,53 +505,61 @@ public class AsyncRpcChannel { byteBuf.writeByte(authMethod.code); } + private void close0(Throwable e) { + List<AsyncCall> toCleanup; + synchronized (pendingCalls) { + if (closed) { + return; + } + closed = true; + toCleanup = new ArrayList<AsyncCall>(pendingCalls.values()); + pendingCalls.clear(); + } + IOException closeException = null; + if (e != null) { + if (e instanceof IOException) { + closeException = (IOException) e; + } else { + closeException = new IOException(e); + } + } + // log the info + if (LOG.isDebugEnabled() && closeException != null) { + LOG.debug(name + ": closing ipc connection to " + address, closeException); + } + if (cleanupTimer != null) { + cleanupTimer.cancel(); + cleanupTimer = null; + } + for (AsyncCall call : toCleanup) { + call.setFailed(closeException != null ? closeException + : new ConnectionClosingException( + "Call id=" + call.id + " on server " + address + " aborted: connection is closing")); + } + channel.disconnect().addListener(ChannelFutureListener.CLOSE); + if (LOG.isDebugEnabled()) { + LOG.debug(name + ": closed"); + } + } + /** * Close connection - * * @param e exception on close */ public void close(final Throwable e) { client.removeConnection(this); // Move closing from the requesting thread to the channel thread - channel.eventLoop().execute(new Runnable() { - @Override - public void run() { - List<AsyncCall> toCleanup; - synchronized (pendingCalls) { - if (closed) { - return; - } - closed = true; - toCleanup = new ArrayList<AsyncCall>(pendingCalls.values()); - pendingCalls.clear(); - } - IOException closeException = null; - if (e != null) { - if (e instanceof IOException) { - closeException = (IOException) e; - } else { - closeException = new IOException(e); - } - } - // log the info - if (LOG.isDebugEnabled() && closeException != null) { - LOG.debug(name + ": closing ipc connection to " + address, closeException); - } - if (cleanupTimer != null) { - cleanupTimer.cancel(); - cleanupTimer = null; - } - for (AsyncCall call : toCleanup) { - call.setFailed(closeException != null ? closeException : new ConnectionClosingException( - "Call id=" + call.id + " on server " + address + " aborted: connection is closing")); - } - channel.disconnect().addListener(ChannelFutureListener.CLOSE); - if (LOG.isDebugEnabled()) { - LOG.debug(name + ": closed"); + if (channel.eventLoop().inEventLoop()) { + close0(e); + } else { + channel.eventLoop().execute(new Runnable() { + @Override + public void run() { + close0(e); } - } - }); + }); + } } /** @@ -592,9 +585,7 @@ public class AsyncRpcChannel { } } if (nextCleanupTaskDelay > 0) { - cleanupTimer = - client.newTimeout(timeoutTask, nextCleanupTaskDelay, - TimeUnit.MILLISECONDS); + cleanupTimer = client.newTimeout(timeoutTask, nextCleanupTaskDelay, TimeUnit.MILLISECONDS); } else { cleanupTimer = null; } @@ -607,7 +598,6 @@ public class AsyncRpcChannel { /** * Check if the connection is alive - * * @return true if alive */ public boolean isAlive() { @@ -616,7 +606,6 @@ public class AsyncRpcChannel { /** * Check if user should authenticate over Kerberos - * * @return true if should be authenticated over Kerberos * @throws java.io.IOException on failure of check */ @@ -624,37 +613,31 @@ public class AsyncRpcChannel { UserGroupInformation loginUser = UserGroupInformation.getLoginUser(); UserGroupInformation currentUser = UserGroupInformation.getCurrentUser(); UserGroupInformation realUser = currentUser.getRealUser(); - return authMethod == AuthMethod.KERBEROS && - loginUser != null && - //Make sure user logged in using Kerberos either keytab or TGT - loginUser.hasKerberosCredentials() && - // relogin only in case it is the login user (e.g. JT) - // or superuser (like oozie). - (loginUser.equals(currentUser) || loginUser.equals(realUser)); + return authMethod == AuthMethod.KERBEROS && loginUser != null && + // Make sure user logged in using Kerberos either keytab or TGT + loginUser.hasKerberosCredentials() && + // relogin only in case it is the login user (e.g. JT) + // or superuser (like oozie). + (loginUser.equals(currentUser) || loginUser.equals(realUser)); } /** - * If multiple clients with the same principal try to connect - * to the same server at the same time, the server assumes a - * replay attack is in progress. This is a feature of kerberos. - * In order to work around this, what is done is that the client - * backs off randomly and tries to initiate the connection - * again. - * The other problem is to do with ticket expiry. To handle that, - * a relogin is attempted. + * If multiple clients with the same principal try to connect to the same server at the same time, + * the server assumes a replay attack is in progress. This is a feature of kerberos. In order to + * work around this, what is done is that the client backs off randomly and tries to initiate the + * connection again. The other problem is to do with ticket expiry. To handle that, a relogin is + * attempted. * <p> - * The retry logic is governed by the {@link #shouldAuthenticateOverKrb} - * method. In case when the user doesn't have valid credentials, we don't - * need to retry (from cache or ticket). In such cases, it is prudent to - * throw a runtime exception when we receive a SaslException from the - * underlying authentication implementation, so there is no retry from - * other high level (for eg, HCM or HBaseAdmin). + * The retry logic is governed by the {@link #shouldAuthenticateOverKrb} method. In case when the + * user doesn't have valid credentials, we don't need to retry (from cache or ticket). In such + * cases, it is prudent to throw a runtime exception when we receive a SaslException from the + * underlying authentication implementation, so there is no retry from other high level (for eg, + * HCM or HBaseAdmin). * </p> - * * @param currRetries retry count - * @param ex exception describing fail - * @param user which is trying to connect - * @throws java.io.IOException if IO fail + * @param ex exception describing fail + * @param user which is trying to connect + * @throws java.io.IOException if IO fail * @throws InterruptedException if thread is interrupted */ private void handleSaslConnectionFailure(final int currRetries, final Throwable ex, @@ -665,7 +648,7 @@ public class AsyncRpcChannel { if (shouldAuthenticateOverKrb()) { if (currRetries < MAX_SASL_RETRIES) { LOG.debug("Exception encountered while connecting to the server : " + ex); - //try re-login + // try re-login if (UserGroupInformation.isLoginKeytabBased()) { UserGroupInformation.getLoginUser().reloginFromKeytab(); } else { @@ -675,23 +658,20 @@ public class AsyncRpcChannel { // Should reconnect return null; } else { - String msg = "Couldn't setup connection for " + - UserGroupInformation.getLoginUser().getUserName() + - " to " + serverPrincipal; + String msg = "Couldn't setup connection for " + + UserGroupInformation.getLoginUser().getUserName() + " to " + serverPrincipal; LOG.warn(msg); throw (IOException) new IOException(msg).initCause(ex); } } else { - LOG.warn("Exception encountered while connecting to " + - "the server : " + ex); + LOG.warn("Exception encountered while connecting to " + "the server : " + ex); } if (ex instanceof RemoteException) { throw (RemoteException) ex; } if (ex instanceof SaslException) { - String msg = "SASL authentication failed." + - " The most likely cause is missing or invalid credentials." + - " Consider 'kinit'."; + String msg = "SASL authentication failed." + + " The most likely cause is missing or invalid credentials." + " Consider 'kinit'."; LOG.fatal(msg, ex); throw new RuntimeException(msg, ex); } @@ -718,7 +698,6 @@ public class AsyncRpcChannel { return false; } - @Override public String toString() { return this.address.toString() + "/" + this.serviceName + "/" + this.ticket; http://git-wip-us.apache.org/repos/asf/hbase/blob/fd5c0934/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AsyncServerResponseHandler.java ---------------------------------------------------------------------- diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AsyncServerResponseHandler.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AsyncServerResponseHandler.java index 8f6c85b..e0c7586 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AsyncServerResponseHandler.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/AsyncServerResponseHandler.java @@ -17,11 +17,6 @@ */ package org.apache.hadoop.hbase.ipc; -import io.netty.buffer.ByteBuf; -import io.netty.buffer.ByteBufInputStream; -import io.netty.channel.ChannelHandlerContext; -import io.netty.channel.ChannelInboundHandlerAdapter; - import java.io.IOException; import org.apache.hadoop.hbase.CellScanner; @@ -32,82 +27,89 @@ import org.apache.hadoop.ipc.RemoteException; import com.google.protobuf.Message; +import io.netty.buffer.ByteBuf; +import io.netty.buffer.ByteBufInputStream; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.SimpleChannelInboundHandler; + /** * Handles Hbase responses */ @InterfaceAudience.Private -public class AsyncServerResponseHandler extends ChannelInboundHandlerAdapter { +public class AsyncServerResponseHandler extends SimpleChannelInboundHandler<ByteBuf> { private final AsyncRpcChannel channel; /** * Constructor - * * @param channel on which this response handler operates */ public AsyncServerResponseHandler(AsyncRpcChannel channel) { this.channel = channel; } - @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { - ByteBuf inBuffer = (ByteBuf) msg; + @Override + protected void channelRead0(ChannelHandlerContext ctx, ByteBuf inBuffer) throws Exception { ByteBufInputStream in = new ByteBufInputStream(inBuffer); int totalSize = inBuffer.readableBytes(); - try { - // Read the header - RPCProtos.ResponseHeader responseHeader = RPCProtos.ResponseHeader.parseDelimitedFrom(in); - int id = responseHeader.getCallId(); - AsyncCall call = channel.removePendingCall(id); - if (call == null) { - // So we got a response for which we have no corresponding 'call' here on the client-side. - // We probably timed out waiting, cleaned up all references, and now the server decides - // to return a response. There is nothing we can do w/ the response at this stage. Clean - // out the wire of the response so its out of the way and we can get other responses on - // this connection. - int readSoFar = IPCUtil.getTotalSizeWhenWrittenDelimited(responseHeader); - int whatIsLeftToRead = totalSize - readSoFar; + // Read the header + RPCProtos.ResponseHeader responseHeader = RPCProtos.ResponseHeader.parseDelimitedFrom(in); + int id = responseHeader.getCallId(); + AsyncCall call = channel.removePendingCall(id); + if (call == null) { + // So we got a response for which we have no corresponding 'call' here on the client-side. + // We probably timed out waiting, cleaned up all references, and now the server decides + // to return a response. There is nothing we can do w/ the response at this stage. Clean + // out the wire of the response so its out of the way and we can get other responses on + // this connection. + int readSoFar = IPCUtil.getTotalSizeWhenWrittenDelimited(responseHeader); + int whatIsLeftToRead = totalSize - readSoFar; - // This is done through a Netty ByteBuf which has different behavior than InputStream. - // It does not return number of bytes read but will update pointer internally and throws an - // exception when too many bytes are to be skipped. - inBuffer.skipBytes(whatIsLeftToRead); - return; - } + // This is done through a Netty ByteBuf which has different behavior than InputStream. + // It does not return number of bytes read but will update pointer internally and throws an + // exception when too many bytes are to be skipped. + inBuffer.skipBytes(whatIsLeftToRead); + return; + } - if (responseHeader.hasException()) { - RPCProtos.ExceptionResponse exceptionResponse = responseHeader.getException(); - RemoteException re = createRemoteException(exceptionResponse); - if (exceptionResponse.getExceptionClassName(). - equals(FatalConnectionException.class.getName())) { - channel.close(re); - } else { - call.setFailed(re); - } + if (responseHeader.hasException()) { + RPCProtos.ExceptionResponse exceptionResponse = responseHeader.getException(); + RemoteException re = createRemoteException(exceptionResponse); + if (exceptionResponse.getExceptionClassName() + .equals(FatalConnectionException.class.getName())) { + channel.close(re); } else { - Message value = null; - // Call may be null because it may have timedout and been cleaned up on this side already - if (call.responseDefaultType != null) { - Message.Builder builder = call.responseDefaultType.newBuilderForType(); - ProtobufUtil.mergeDelimitedFrom(builder, in); - value = builder.build(); - } - CellScanner cellBlockScanner = null; - if (responseHeader.hasCellBlockMeta()) { - int size = responseHeader.getCellBlockMeta().getLength(); - byte[] cellBlock = new byte[size]; - inBuffer.readBytes(cellBlock, 0, cellBlock.length); - cellBlockScanner = channel.client.createCellScanner(cellBlock); - } - call.setSuccess(value, cellBlockScanner); - call.callStats.setResponseSizeBytes(totalSize); + call.setFailed(re); + } + } else { + Message value = null; + // Call may be null because it may have timedout and been cleaned up on this side already + if (call.responseDefaultType != null) { + Message.Builder builder = call.responseDefaultType.newBuilderForType(); + ProtobufUtil.mergeDelimitedFrom(builder, in); + value = builder.build(); + } + CellScanner cellBlockScanner = null; + if (responseHeader.hasCellBlockMeta()) { + int size = responseHeader.getCellBlockMeta().getLength(); + byte[] cellBlock = new byte[size]; + inBuffer.readBytes(cellBlock, 0, cellBlock.length); + cellBlockScanner = channel.client.createCellScanner(cellBlock); } - } catch (IOException e) { - // Treat this as a fatal condition and close this connection - channel.close(e); - } finally { - inBuffer.release(); + call.setSuccess(value, cellBlockScanner); + call.callStats.setResponseSizeBytes(totalSize); } } + @Override + public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { + channel.close(cause); + } + + @Override + public void channelInactive(ChannelHandlerContext ctx) throws Exception { + channel.close(new IOException("connection closed")); + } + /** * @param e Proto exception * @return RemoteException made from passed <code>e</code> @@ -118,7 +120,7 @@ public class AsyncServerResponseHandler extends ChannelInboundHandlerAdapter { return e.hasHostname() ? // If a hostname then add it to the RemoteWithExtrasException new RemoteWithExtrasException(innerExceptionClassName, e.getStackTrace(), e.getHostname(), - e.getPort(), doNotRetry) : - new RemoteWithExtrasException(innerExceptionClassName, e.getStackTrace(), doNotRetry); + e.getPort(), doNotRetry) + : new RemoteWithExtrasException(innerExceptionClassName, e.getStackTrace(), doNotRetry); } } http://git-wip-us.apache.org/repos/asf/hbase/blob/fd5c0934/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 e8da9ee..69c8fe2 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 @@ -57,7 +57,6 @@ import org.apache.hadoop.hbase.util.Bytes; import org.apache.hadoop.hbase.util.Pair; import org.apache.hadoop.io.compress.GzipCodec; import org.apache.hadoop.util.StringUtils; -import org.apache.http.ConnectionClosedException; import org.junit.Assert; import org.junit.Test; @@ -320,7 +319,7 @@ public abstract class AbstractTestIPC { md.getOutputType().toProto(), User.getCurrent(), address, new MetricsConnection.CallStats()); fail("RPC should have failed because it exceeds max request size"); - } catch(ConnectionClosingException | ConnectionClosedException ex) { + } catch(IOException ex) { // pass } } finally {
