You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by la...@apache.org on 2015/02/08 08:37:54 UTC
hbase git commit: HBASE-12968 [0.94]SecureServer should not ignore
CallQueueSize. (hongyu bi)
Repository: hbase
Updated Branches:
refs/heads/0.94 a0b773824 -> 25d4d266f
HBASE-12968 [0.94]SecureServer should not ignore CallQueueSize. (hongyu bi)
Project: http://git-wip-us.apache.org/repos/asf/hbase/repo
Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/25d4d266
Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/25d4d266
Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/25d4d266
Branch: refs/heads/0.94
Commit: 25d4d266f3c07e2af3663b8e806a0bb6b015522e
Parents: a0b7738
Author: Lars Hofhansl <la...@apache.org>
Authored: Sat Feb 7 23:39:08 2015 -0800
Committer: Lars Hofhansl <la...@apache.org>
Committed: Sat Feb 7 23:39:08 2015 -0800
----------------------------------------------------------------------
.../org/apache/hadoop/hbase/ipc/SecureServer.java | 18 ++++++++++++++++--
.../org/apache/hadoop/hbase/ipc/HBaseServer.java | 4 ++--
2 files changed, 18 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hbase/blob/25d4d266/security/src/main/java/org/apache/hadoop/hbase/ipc/SecureServer.java
----------------------------------------------------------------------
diff --git a/security/src/main/java/org/apache/hadoop/hbase/ipc/SecureServer.java b/security/src/main/java/org/apache/hadoop/hbase/ipc/SecureServer.java
index e630125..c5fe8c1 100644
--- a/security/src/main/java/org/apache/hadoop/hbase/ipc/SecureServer.java
+++ b/security/src/main/java/org/apache/hadoop/hbase/ipc/SecureServer.java
@@ -612,15 +612,29 @@ public abstract class SecureServer extends HBaseServer {
DataInputStream dis =
new DataInputStream(new ByteArrayInputStream(buf));
int id = dis.readInt(); // try to read an id
+ long callSize = buf.length;
if (LOG.isTraceEnabled()) {
LOG.trace(" got #" + id);
}
+ // Enforcing the call queue size, this triggers a retry in the client
+ if ((callSize + callQueueSize.get()) > maxQueueSize) {
+ final SecureCall callTooBig =
+ new SecureCall(id, null, this, responder, callSize);
+ ByteArrayOutputStream responseBuffer = new ByteArrayOutputStream();
+ setupResponse(responseBuffer, callTooBig, Status.FATAL, null,
+ IOException.class.getName(),
+ "Call queue is full, is ipc.server.max.callqueue.size too small?");
+ responder.doRespond(callTooBig);
+ return;
+ }
+
Writable param = ReflectionUtils.newInstance(paramClass, conf); // read param
param.readFields(dis);
- SecureCall call = new SecureCall(id, param, this, responder, buf.length);
+ SecureCall call = new SecureCall(id, param, this, responder, callSize);
+ callQueueSize.add(callSize);
if (priorityCallQueue != null && getQosLevel(param) > highPriorityLevel) {
priorityCallQueue.put(call);
@@ -757,4 +771,4 @@ public abstract class SecureServer extends HBaseServer {
protocol, getConf(), addr);
}
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/hbase/blob/25d4d266/src/main/java/org/apache/hadoop/hbase/ipc/HBaseServer.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/hadoop/hbase/ipc/HBaseServer.java b/src/main/java/org/apache/hadoop/hbase/ipc/HBaseServer.java
index 261ce4a..417e2a2 100644
--- a/src/main/java/org/apache/hadoop/hbase/ipc/HBaseServer.java
+++ b/src/main/java/org/apache/hadoop/hbase/ipc/HBaseServer.java
@@ -204,7 +204,7 @@ public abstract class HBaseServer implements RpcServer {
protected Configuration conf;
private int maxQueueLength;
- private int maxQueueSize;
+ protected int maxQueueSize;
protected int socketSendBufferSize;
protected final boolean tcpNoDelay; // if T then disable Nagle's Algorithm
protected final boolean tcpKeepAlive; // if T then use keepalives
@@ -1606,7 +1606,7 @@ public abstract class HBaseServer implements RpcServer {
* @param error error message, if the call failed
* @throws IOException
*/
- private void setupResponse(ByteArrayOutputStream response,
+ protected void setupResponse(ByteArrayOutputStream response,
Call call, Status status,
Writable rv, String errorClass, String error)
throws IOException {