HBASE-16752 Return error back to rpc client on exceeding rpc request size limit.
Signed-off-by: Gary Helmling <garyh@apache.org>
This commit is contained in:
parent
a97aef5163
commit
33e89fa9cf
|
@ -63,7 +63,7 @@ public final class ClientExceptionsUtil {
|
|||
|| cur instanceof RegionTooBusyException || cur instanceof ThrottlingException
|
||||
|| cur instanceof MultiActionResultTooLarge || cur instanceof RetryImmediatelyException
|
||||
|| cur instanceof CallQueueTooBigException || cur instanceof CallDroppedException
|
||||
|| cur instanceof NotServingRegionException);
|
||||
|| cur instanceof NotServingRegionException || cur instanceof RequestTooBigException);
|
||||
}
|
||||
|
||||
|
||||
|
|
|
@ -0,0 +1,43 @@
|
|||
/**
|
||||
* Licensed to the Apache Software Foundation (ASF) under one
|
||||
* or more contributor license agreements. See the NOTICE file
|
||||
* distributed with this work for additional information
|
||||
* regarding copyright ownership. The ASF licenses this file
|
||||
* to you under the Apache License, Version 2.0 (the
|
||||
* "License"); you may not use this file except in compliance
|
||||
* with the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.apache.hadoop.hbase.exceptions;
|
||||
|
||||
import org.apache.hadoop.hbase.DoNotRetryIOException;
|
||||
import org.apache.hadoop.hbase.classification.InterfaceAudience;
|
||||
import org.apache.hadoop.hbase.classification.InterfaceStability;
|
||||
|
||||
/**
|
||||
* Thrown when the size of the rpc request received by the server is too large.
|
||||
*
|
||||
* On receiving such an exception, the client does not retry the offending rpc.
|
||||
*/
|
||||
@InterfaceAudience.Public
|
||||
@InterfaceStability.Evolving
|
||||
public class RequestTooBigException extends DoNotRetryIOException {
|
||||
|
||||
private static final long serialVersionUID = -1593339239809586516L;
|
||||
|
||||
public RequestTooBigException() {
|
||||
super();
|
||||
}
|
||||
|
||||
public RequestTooBigException(String message) {
|
||||
super(message);
|
||||
}
|
||||
}
|
|
@ -24,6 +24,7 @@ import java.io.ByteArrayInputStream;
|
|||
import java.io.ByteArrayOutputStream;
|
||||
import java.io.DataOutputStream;
|
||||
import java.io.IOException;
|
||||
import java.io.InputStream;
|
||||
import java.net.BindException;
|
||||
import java.net.InetAddress;
|
||||
import java.net.InetSocketAddress;
|
||||
|
@ -85,6 +86,7 @@ import org.apache.hadoop.hbase.client.VersionInfoUtil;
|
|||
import org.apache.hadoop.hbase.codec.Codec;
|
||||
import org.apache.hadoop.hbase.conf.ConfigurationObserver;
|
||||
import org.apache.hadoop.hbase.exceptions.RegionMovedException;
|
||||
import org.apache.hadoop.hbase.exceptions.RequestTooBigException;
|
||||
import org.apache.hadoop.hbase.io.BoundedByteBufferPool;
|
||||
import org.apache.hadoop.hbase.io.ByteBufferInputStream;
|
||||
import org.apache.hadoop.hbase.io.ByteBufferOutputStream;
|
||||
|
@ -263,6 +265,9 @@ public class RpcServer implements RpcServerInterface, ConfigurationObserver {
|
|||
protected HBaseRPCErrorHandler errorHandler = null;
|
||||
|
||||
static final String MAX_REQUEST_SIZE = "hbase.ipc.max.request.size";
|
||||
private static final RequestTooBigException REQUEST_TOO_BIG_EXCEPTION =
|
||||
new RequestTooBigException();
|
||||
|
||||
private static final String WARN_RESPONSE_TIME = "hbase.ipc.warn.response.time";
|
||||
private static final String WARN_RESPONSE_SIZE = "hbase.ipc.warn.response.size";
|
||||
|
||||
|
@ -1653,9 +1658,47 @@ public class RpcServer implements RpcServerInterface, ConfigurationObserver {
|
|||
}
|
||||
|
||||
if (dataLength > maxRequestSize) {
|
||||
throw new DoNotRetryIOException("RPC data length of " + dataLength + " received from "
|
||||
+ getHostAddress() + " is greater than max allowed " + maxRequestSize + ". Set \""
|
||||
+ MAX_REQUEST_SIZE + "\" on server to override this limit (not recommended)");
|
||||
String msg = "RPC data length of " + dataLength + " received from "
|
||||
+ getHostAddress() + " is greater than max allowed "
|
||||
+ maxRequestSize + ". Set \"" + MAX_REQUEST_SIZE
|
||||
+ "\" on server to override this limit (not recommended)";
|
||||
LOG.warn(msg);
|
||||
|
||||
if (connectionHeaderRead && connectionPreambleRead) {
|
||||
incRpcCount();
|
||||
// Construct InputStream for the non-blocking SocketChannel
|
||||
// We need the InputStream because we want to read only the request header
|
||||
// instead of the whole rpc.
|
||||
final ByteBuffer buf = ByteBuffer.allocate(1);
|
||||
InputStream is = new InputStream() {
|
||||
@Override
|
||||
public int read() throws IOException {
|
||||
channelRead(channel, buf);
|
||||
buf.flip();
|
||||
int x = buf.get();
|
||||
buf.flip();
|
||||
return x;
|
||||
}
|
||||
};
|
||||
CodedInputStream cis = CodedInputStream.newInstance(is);
|
||||
int headerSize = cis.readRawVarint32();
|
||||
Message.Builder builder = RequestHeader.newBuilder();
|
||||
ProtobufUtil.mergeFrom(builder, cis, headerSize);
|
||||
RequestHeader header = (RequestHeader) builder.build();
|
||||
|
||||
// Notify the client about the offending request
|
||||
Call reqTooBig = new Call(header.getCallId(), this.service, null, null, null,
|
||||
null, this, responder, 0, null, this.addr,0);
|
||||
metrics.exception(REQUEST_TOO_BIG_EXCEPTION);
|
||||
setupResponse(null, reqTooBig, REQUEST_TOO_BIG_EXCEPTION, msg);
|
||||
// We are going to close the connection, make sure we process the response
|
||||
// before that. In rare case when this fails, we still close the connection.
|
||||
responseWriteLock.lock();
|
||||
responder.processResponse(reqTooBig);
|
||||
responseWriteLock.unlock();
|
||||
}
|
||||
// Close the connection
|
||||
return -1;
|
||||
}
|
||||
|
||||
data = ByteBuffer.allocate(dataLength);
|
||||
|
|
|
@ -300,15 +300,18 @@ public abstract class AbstractTestIPC {
|
|||
@Test
|
||||
public void testRpcMaxRequestSize() throws IOException, InterruptedException {
|
||||
Configuration conf = new Configuration(CONF);
|
||||
conf.setInt(RpcServer.MAX_REQUEST_SIZE, 100);
|
||||
conf.setInt(RpcServer.MAX_REQUEST_SIZE, 1000);
|
||||
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();
|
||||
StringBuilder message = new StringBuilder(1200);
|
||||
for (int i = 0; i < 200; i++) {
|
||||
message.append("hello.");
|
||||
}
|
||||
// set total RPC size bigger than 1000 bytes
|
||||
EchoRequestProto param = EchoRequestProto.newBuilder().setMessage(message.toString()).build();
|
||||
InetSocketAddress address = rpcServer.getListenerAddress();
|
||||
if (address == null) {
|
||||
throw new IOException("Listener channel is closed");
|
||||
|
@ -319,8 +322,10 @@ 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(IOException ex) {
|
||||
// pass
|
||||
} catch(IOException e) {
|
||||
LOG.info("Caught expected exception: " + e);
|
||||
assertTrue(e.toString(),
|
||||
StringUtils.stringifyException(e).contains("RequestTooBigException"));
|
||||
}
|
||||
} finally {
|
||||
rpcServer.stop();
|
||||
|
|
Loading…
Reference in New Issue