From 6d080762ef795adf02dd0ab236c4b3eb73e19a91 Mon Sep 17 00:00:00 2001 From: Nihal Jain Date: Fri, 27 Apr 2018 14:13:57 +0530 Subject: [PATCH] HBASE-20499 Replication/Priority executors can use specific max queue length as default value instead of general maxQueueLength Signed-off-by: tedyu --- .../org/apache/hadoop/hbase/ipc/RpcExecutor.java | 8 ++++++-- .../apache/hadoop/hbase/ipc/RpcScheduler.java | 2 ++ .../hadoop/hbase/ipc/SimpleRpcScheduler.java | 16 +++++++++++----- 3 files changed, 19 insertions(+), 7 deletions(-) diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/RpcExecutor.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/RpcExecutor.java index 7470758032e..f63b243f188 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/RpcExecutor.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/RpcExecutor.java @@ -486,8 +486,12 @@ public abstract class RpcExecutor { */ public void resizeQueues(Configuration conf) { String configKey = RpcScheduler.IPC_SERVER_MAX_CALLQUEUE_LENGTH; - if (name != null && name.toLowerCase(Locale.ROOT).contains("priority")) { - configKey = RpcScheduler.IPC_SERVER_PRIORITY_MAX_CALLQUEUE_LENGTH; + if (name != null) { + if (name.toLowerCase(Locale.ROOT).contains("priority")) { + configKey = RpcScheduler.IPC_SERVER_PRIORITY_MAX_CALLQUEUE_LENGTH; + } else if (name.toLowerCase(Locale.ROOT).contains("replication")) { + configKey = RpcScheduler.IPC_SERVER_REPLICATION_MAX_CALLQUEUE_LENGTH; + } } currentQueueLimit = conf.getInt(configKey, currentQueueLimit); } diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/RpcScheduler.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/RpcScheduler.java index be54e54666b..e1fe397d383 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/RpcScheduler.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/RpcScheduler.java @@ -35,6 +35,8 @@ public abstract class RpcScheduler { "hbase.ipc.server.max.callqueue.length"; public static final String IPC_SERVER_PRIORITY_MAX_CALLQUEUE_LENGTH = "hbase.ipc.server.priority.max.callqueue.length"; + public static final String IPC_SERVER_REPLICATION_MAX_CALLQUEUE_LENGTH = + "hbase.ipc.server.replication.max.callqueue.length"; /** Exposes runtime information of a {@code RpcServer} that a {@code RpcScheduler} may need. */ public static abstract class Context { diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/SimpleRpcScheduler.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/SimpleRpcScheduler.java index 47c1498cfb8..725a93a6c93 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/SimpleRpcScheduler.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/SimpleRpcScheduler.java @@ -65,8 +65,11 @@ public class SimpleRpcScheduler extends RpcScheduler implements ConfigurationObs int maxQueueLength = conf.getInt(RpcScheduler.IPC_SERVER_MAX_CALLQUEUE_LENGTH, handlerCount * RpcServer.DEFAULT_MAX_CALLQUEUE_LENGTH_PER_HANDLER); - int maxPriorityQueueLength = - conf.getInt(RpcScheduler.IPC_SERVER_PRIORITY_MAX_CALLQUEUE_LENGTH, maxQueueLength); + int maxPriorityQueueLength = conf.getInt(RpcScheduler.IPC_SERVER_PRIORITY_MAX_CALLQUEUE_LENGTH, + priorityHandlerCount * RpcServer.DEFAULT_MAX_CALLQUEUE_LENGTH_PER_HANDLER); + int maxReplicationQueueLength = + conf.getInt(RpcScheduler.IPC_SERVER_REPLICATION_MAX_CALLQUEUE_LENGTH, + replicationHandlerCount * RpcServer.DEFAULT_MAX_CALLQUEUE_LENGTH_PER_HANDLER); this.priority = priority; this.highPriorityLevel = highPriorityLevel; @@ -94,9 +97,12 @@ public class SimpleRpcScheduler extends RpcScheduler implements ConfigurationObs this.priorityExecutor = priorityHandlerCount > 0 ? new FastPathBalancedQueueRpcExecutor( "priority.FPBQ", priorityHandlerCount, RpcExecutor.CALL_QUEUE_TYPE_FIFO_CONF_VALUE, maxPriorityQueueLength, priority, conf, abortable) : null; - this.replicationExecutor = replicationHandlerCount > 0 ? new FastPathBalancedQueueRpcExecutor( - "replication.FPBQ", replicationHandlerCount, RpcExecutor.CALL_QUEUE_TYPE_FIFO_CONF_VALUE, - maxQueueLength, priority, conf, abortable) : null; + this.replicationExecutor = + replicationHandlerCount > 0 + ? new FastPathBalancedQueueRpcExecutor("replication.FPBQ", replicationHandlerCount, + RpcExecutor.CALL_QUEUE_TYPE_FIFO_CONF_VALUE, maxReplicationQueueLength, priority, + conf, abortable) + : null; }