HBASE-27144 Improvement bulkload rpc

Author: SiCheng-Zheng

hbase-common/src/main/java/org/apache/hadoop/hbase/HConstants.java

@@ -1063,6 +1063,13 @@ public enum OperationStatusCode {
   public static final String REGION_SERVER_REPLICATION_HANDLER_COUNT =
     "hbase.regionserver.replication.handler.count";
   public static final int DEFAULT_REGION_SERVER_REPLICATION_HANDLER_COUNT = 3;
+  public static final String REGION_SERVER_BULKLOAD_HANDLER_COUNT =
+    "hbase.regionserver.bulkload.handler.count";
+  public static final int DEFAULT_REGION_SERVER_BULKLOAD_HANDLER_COUNT = 1;
+  public static final String HBASE_REGIONSERVER_BULKLOAD_INDEPENDENCE_QUEUE_ENABLED =
+    "hbase.regionserver.bulkload.independence.queue.enabled";
+  public static final boolean DEFAULT_HBASE_REGIONSERVER_BULKLOAD_INDEPENDENCE_QUEUE_ENABLED =
+    false;
   // Meta Transition handlers to deal with meta ReportRegionStateTransitionRequest. Meta transition
   // should be dealt with in a separate handler in case blocking other region's transition.
   public static final String MASTER_META_TRANSITION_HANDLER_COUNT =
@@ -1135,10 +1142,11 @@ public enum OperationStatusCode {
    * QOS attributes: these attributes are used to demarcate RPC call processing by different set of
    * handlers. For example, HIGH_QOS tagged methods are handled by high priority handlers.
    */
-  // normal_QOS < replication_QOS < replay_QOS < QOS_threshold < admin_QOS < high_QOS < meta_QOS
+  // normal_QOS < BULKLOAD_QOS < replication_QOS < replay_QOS < QOS_threshold < admin_QOS < high_QOS < meta_QOS
   public static final int PRIORITY_UNSET = -1;
   public static final int NORMAL_QOS = 0;
   public static final int REPLICATION_QOS = 5;
+  public static final int BULKLOAD_QOS = 4;
   /**
    * @deprecated since 3.0.0, will be removed in 4.0.0. DLR has been purged for a long time and
    *             region replication has its own 'replay' method.

hbase-hadoop-compat/src/main/java/org/apache/hadoop/hbase/ipc/MetricsHBaseServerSource.java

@@ -56,6 +56,9 @@ public interface MetricsHBaseServerSource extends ExceptionTrackingSource {
   String METAPRIORITY_QUEUE_NAME = "numCallsInMetaPriorityQueue";
   String REPLICATION_QUEUE_NAME = "numCallsInReplicationQueue";
   String REPLICATION_QUEUE_DESC = "Number of calls in the replication call queue waiting to be run";
+  String BULKLOAD_QUEUE_NAME = "numCallsInBulkLoadQueue";
+  String BULKLOAD_QUEUE_DESC =
+    "Number of calls in the bulkload call queue waiting to be run";
   String PRIORITY_QUEUE_DESC = "Number of calls in the priority call queue waiting to be run";
   String METAPRIORITY_QUEUE_DESC = "Number of calls in the priority call queue waiting to be run";
   String WRITE_QUEUE_NAME = "numCallsInWriteQueue";
@@ -77,6 +80,8 @@ public interface MetricsHBaseServerSource extends ExceptionTrackingSource {
   String NUM_ACTIVE_PRIORITY_HANDLER_DESC = "Number of active priority rpc handlers.";
   String NUM_ACTIVE_REPLICATION_HANDLER_NAME = "numActiveReplicationHandler";
   String NUM_ACTIVE_REPLICATION_HANDLER_DESC = "Number of active replication rpc handlers.";
+  String NUM_ACTIVE_BULKLOAD_HANDLER_NAME = "numActiveBulkLoadHandler";
+  String NUM_ACTIVE_BULKLOAD_HANDLER_DESC = "Number of active bulkload rpc handlers.";
   String NUM_ACTIVE_WRITE_HANDLER_NAME = "numActiveWriteHandler";
   String NUM_ACTIVE_WRITE_HANDLER_DESC = "Number of active write rpc handlers.";
   String NUM_ACTIVE_READ_HANDLER_NAME = "numActiveReadHandler";

hbase-hadoop-compat/src/main/java/org/apache/hadoop/hbase/ipc/MetricsHBaseServerSourceImpl.java

@@ -143,6 +143,8 @@ public void getMetrics(MetricsCollector metricsCollector, boolean all) {
           wrapper.getGeneralQueueLength())
         .addGauge(Interns.info(REPLICATION_QUEUE_NAME, REPLICATION_QUEUE_DESC),
           wrapper.getReplicationQueueLength())
+        .addGauge(Interns.info(BULKLOAD_QUEUE_NAME, BULKLOAD_QUEUE_DESC),
+          wrapper.getBulkLoadQueueLength())
         .addGauge(Interns.info(PRIORITY_QUEUE_NAME, PRIORITY_QUEUE_DESC),
           wrapper.getPriorityQueueLength())
         .addGauge(Interns.info(METAPRIORITY_QUEUE_NAME, METAPRIORITY_QUEUE_DESC),
@@ -163,6 +165,9 @@ public void getMetrics(MetricsCollector metricsCollector, boolean all) {
         .addCounter(Interns.info(NUM_LIFO_MODE_SWITCHES_NAME, NUM_LIFO_MODE_SWITCHES_DESC),
           wrapper.getNumLifoModeSwitches())
         .addGauge(Interns.info(WRITE_QUEUE_NAME, WRITE_QUEUE_DESC), wrapper.getWriteQueueLength())
+        .addGauge(
+          Interns.info(NUM_ACTIVE_BULKLOAD_HANDLER_NAME, NUM_ACTIVE_BULKLOAD_HANDLER_DESC),
+          wrapper.getActiveBulkLoadRpcHandlerCount())
         .addGauge(Interns.info(READ_QUEUE_NAME, READ_QUEUE_DESC), wrapper.getReadQueueLength())
         .addGauge(Interns.info(SCAN_QUEUE_NAME, SCAN_QUEUE_DESC), wrapper.getScanQueueLength())
         .addGauge(Interns.info(NUM_ACTIVE_WRITE_HANDLER_NAME, NUM_ACTIVE_WRITE_HANDLER_DESC),

hbase-hadoop-compat/src/main/java/org/apache/hadoop/hbase/ipc/MetricsHBaseServerWrapper.java

@@ -27,6 +27,8 @@ public interface MetricsHBaseServerWrapper {
 
   int getReplicationQueueLength();
 
+  int getBulkLoadQueueLength();
+
   int getPriorityQueueLength();
 
   int getMetaPriorityQueueLength();
@@ -41,6 +43,8 @@ public interface MetricsHBaseServerWrapper {
 
   int getActiveReplicationRpcHandlerCount();
 
+  int getActiveBulkLoadRpcHandlerCount();
+
   int getActiveMetaPriorityRpcHandlerCount();
 
   long getNumGeneralCallsDropped();

hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/FifoRpcScheduler.java

@@ -130,6 +130,10 @@ public int getReplicationQueueLength() {
     return 0;
   }
 
+  @Override public int getBulkLoadQueueLength() {
+    return 0;
+  }
+
   @Override
   public int getActiveRpcHandlerCount() {
     return executor.getActiveCount();
@@ -150,6 +154,10 @@ public int getActiveReplicationRpcHandlerCount() {
     return 0;
   }
 
+  @Override public int getActiveBulkLoadRpcHandlerCount() {
+    return 0;
+  }
+
   @Override
   public int getActiveMetaPriorityRpcHandlerCount() {
     return 0;

hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/MetricsHBaseServerWrapperImpl.java

@@ -57,6 +57,13 @@ public int getReplicationQueueLength() {
     return server.getScheduler().getReplicationQueueLength();
   }
 
+  @Override public int getBulkLoadQueueLength() {
+    if (!isServerStarted() || this.server.getScheduler() == null) {
+      return 0;
+    }
+    return server.getScheduler().getBulkLoadQueueLength();
+  }
+
   @Override
   public int getPriorityQueueLength() {
     if (!isServerStarted() || this.server.getScheduler() == null) {
@@ -121,6 +128,13 @@ public int getActiveReplicationRpcHandlerCount() {
     return server.getScheduler().getActiveReplicationRpcHandlerCount();
   }
 
+  @Override public int getActiveBulkLoadRpcHandlerCount() {
+    if (!isServerStarted() || this.server.getScheduler() == null) {
+      return 0;
+    }
+    return server.getScheduler().getActiveBulkLoadRpcHandlerCount();
+  }
+
   @Override
   public long getNumGeneralCallsDropped() {
     if (!isServerStarted() || this.server.getScheduler() == null) {

hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/RpcScheduler.java

@@ -35,6 +35,8 @@ public abstract class RpcScheduler {
     "hbase.ipc.server.priority.max.callqueue.length";
   public static final String IPC_SERVER_REPLICATION_MAX_CALLQUEUE_LENGTH =
     "hbase.ipc.server.replication.max.callqueue.length";
+  public static final String IPC_SERVER_BULKLOAD_MAX_CALLQUEUE_LENGTH =
+    "hbase.ipc.server.bulkload.max.callqueue.length";
 
   /** Exposes runtime information of a {@code RpcServer} that a {@code RpcScheduler} may need. */
   public static abstract class Context {
@@ -78,6 +80,9 @@ public static abstract class Context {
   /** Retrieves length of the replication queue for metrics. */
   public abstract int getReplicationQueueLength();
 
+  /** Retrieves length of the bulkload queue for metrics. */
+  public abstract int getBulkLoadQueueLength();
+
   /** Retrieves the total number of active handler. */
   public abstract int getActiveRpcHandlerCount();
 
@@ -93,6 +98,9 @@ public static abstract class Context {
   /** Retrieves the number of active replication handler. */
   public abstract int getActiveReplicationRpcHandlerCount();
 
+  /** Retrieves the number of active bulkload handler. */
+  public abstract int getActiveBulkLoadRpcHandlerCount();
+
   /**
    * If CoDel-based RPC executors are used, retrieves the number of Calls that were dropped from
    * general queue because RPC executor is under high load; returns 0 otherwise.

hbase-server/src/main/java/org/apache/hadoop/hbase/ipc/SimpleRpcScheduler.java

@@ -49,11 +49,38 @@ public class SimpleRpcScheduler extends RpcScheduler implements ConfigurationObs
    */
   private final RpcExecutor metaTransitionExecutor;
 
+  private RpcExecutor bulkloadExecutor;
+
   /** What level a high priority call is at. */
   private final int highPriorityLevel;
 
   private Abortable abortable = null;
 
+  /**
+   * @param conf conf
+   * @param handlerCount the number of handler threads that will be used to process calls
+   * @param priorityHandlerCount How many threads for priority handling.
+   * @param replicationHandlerCount How many threads for replication handling.
+   * @param metaTransitionHandler How many threads for meta handling.
+   * @param bulkLoadHandlerCount How many threads for bulkload handling
+   * @param highPriorityLevel high Priority Level
+   * @param priority Function to extract request priority.
+   */
+  public SimpleRpcScheduler(Configuration conf, int handlerCount, int priorityHandlerCount,
+    int replicationHandlerCount, int metaTransitionHandler, int bulkLoadHandlerCount,
+    PriorityFunction priority, Abortable server, int highPriorityLevel) {
+    this(conf, handlerCount, priorityHandlerCount, replicationHandlerCount,
+      metaTransitionHandler, priority, server, highPriorityLevel);
+    int maxBulkLoadQueueLength =
+      conf.getInt(RpcScheduler.IPC_SERVER_BULKLOAD_MAX_CALLQUEUE_LENGTH,
+        bulkLoadHandlerCount * RpcServer.DEFAULT_MAX_CALLQUEUE_LENGTH_PER_HANDLER);
+    this.bulkloadExecutor = bulkLoadHandlerCount > 0 ?
+      new FastPathBalancedQueueRpcExecutor("bulkLoad.FPBQ", bulkLoadHandlerCount,
+        RpcExecutor.CALL_QUEUE_TYPE_FIFO_CONF_VALUE, maxBulkLoadQueueLength, priority, conf,
+        abortable) :
+      null;
+  }
+
   /**
    * n * @param handlerCount the number of handler threads that will be used to process calls
    * @param priorityHandlerCount    How many threads for priority handling.
@@ -130,6 +157,14 @@ public SimpleRpcScheduler(Configuration conf, int handlerCount, int priorityHand
       highPriorityLevel);
   }
 
+  public SimpleRpcScheduler(Configuration conf, int handlerCount, int priorityHandlerCount,
+    int replicationHandlerCount, int bulkLoadHandlerCount,
+    PriorityFunction priority, int highPriorityLevel) {
+    this(conf, handlerCount, priorityHandlerCount, replicationHandlerCount,
+      0, bulkLoadHandlerCount, priority, null,
+      highPriorityLevel);
+  }
+
   /**
    * Resize call queues;
    * @param conf new configuration
@@ -173,6 +208,9 @@ public void start() {
     if (metaTransitionExecutor != null) {
       metaTransitionExecutor.start(port);
     }
+    if (bulkloadExecutor != null) {
+      bulkloadExecutor.start(port);
+    }
 
   }
 
@@ -188,6 +226,9 @@ public void stop() {
     if (metaTransitionExecutor != null) {
       metaTransitionExecutor.stop();
     }
+    if (bulkloadExecutor != null) {
+      bulkloadExecutor.stop();
+    }
 
   }
 
@@ -208,6 +249,8 @@ public boolean dispatch(CallRunner callTask) {
       return priorityExecutor.dispatch(callTask);
     } else if (replicationExecutor != null && level == HConstants.REPLICATION_QOS) {
       return replicationExecutor.dispatch(callTask);
+    } else if (bulkloadExecutor != null && level == HConstants.BULKLOAD_QOS) {
+      return bulkloadExecutor.dispatch(callTask);
     } else {
       return callExecutor.dispatch(callTask);
     }
@@ -233,10 +276,15 @@ public int getReplicationQueueLength() {
     return replicationExecutor == null ? 0 : replicationExecutor.getQueueLength();
   }
 
+  @Override public int getBulkLoadQueueLength() {
+    return bulkloadExecutor == null ? 0 : bulkloadExecutor.getQueueLength();
+  }
+
   @Override
   public int getActiveRpcHandlerCount() {
     return callExecutor.getActiveHandlerCount() + getActivePriorityRpcHandlerCount()
-      + getActiveReplicationRpcHandlerCount() + getActiveMetaPriorityRpcHandlerCount();
+      + getActiveReplicationRpcHandlerCount() + getActiveMetaPriorityRpcHandlerCount()
+      + getActiveBulkLoadRpcHandlerCount();
   }
 
   @Override
@@ -259,6 +307,10 @@ public int getActiveReplicationRpcHandlerCount() {
     return (replicationExecutor == null ? 0 : replicationExecutor.getActiveHandlerCount());
   }
 
+  @Override public int getActiveBulkLoadRpcHandlerCount() {
+    return bulkloadExecutor == null ? 0 : bulkloadExecutor.getActiveHandlerCount();
+  }
+
   @Override
   public long getNumGeneralCallsDropped() {
     return callExecutor.getNumGeneralCallsDropped();
@@ -330,6 +382,13 @@ public CallQueueInfo getCallQueueInfo() {
       callQueueInfo.setCallMethodSize(queueName, metaTransitionExecutor.getCallQueueSizeSummary());
     }
 
+    if (null != bulkloadExecutor) {
+      queueName = "BulkLoad Queue";
+      callQueueInfo.setCallMethodCount(queueName,
+        bulkloadExecutor.getCallQueueCountsSummary());
+      callQueueInfo.setCallMethodSize(queueName, bulkloadExecutor.getCallQueueSizeSummary());
+    }
+
     return callQueueInfo;
   }
 

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSAnnotationReadingPriorityFunction.java

@@ -38,6 +38,7 @@
 import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.GetRequest;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.MutateRequest;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.ScanRequest;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.BulkLoadHFileRequest;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos.RegionSpecifier;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.RPCProtos.RequestHeader;
 
@@ -96,6 +97,10 @@ protected int getBasePriority(RequestHeader header, Message param) {
     if (header.hasPriority()) {
       return header.getPriority();
     }
+    if (param instanceof BulkLoadHFileRequest) {
+      return HConstants.BULKLOAD_QOS;
+    }
+
     String cls = param.getClass().getName();
     Class<? extends Message> rpcArgClass = argumentToClassMap.get(cls);
     RegionSpecifier regionSpecifier = null;

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/SimpleRpcSchedulerFactory.java

@@ -37,13 +37,29 @@ public class SimpleRpcSchedulerFactory implements RpcSchedulerFactory {
   public RpcScheduler create(Configuration conf, PriorityFunction priority, Abortable server) {
     int handlerCount = conf.getInt(HConstants.REGION_SERVER_HANDLER_COUNT,
       HConstants.DEFAULT_REGION_SERVER_HANDLER_COUNT);
-    return new SimpleRpcScheduler(conf, handlerCount,
-      conf.getInt(HConstants.REGION_SERVER_HIGH_PRIORITY_HANDLER_COUNT,
-        HConstants.DEFAULT_REGION_SERVER_HIGH_PRIORITY_HANDLER_COUNT),
-      conf.getInt(HConstants.REGION_SERVER_REPLICATION_HANDLER_COUNT,
-        HConstants.DEFAULT_REGION_SERVER_REPLICATION_HANDLER_COUNT),
-      conf.getInt(HConstants.MASTER_META_TRANSITION_HANDLER_COUNT,
-        HConstants.MASTER__META_TRANSITION_HANDLER_COUNT_DEFAULT),
-      priority, server, HConstants.QOS_THRESHOLD);
+    boolean bulkLoadIndependenceQueueEnabled = conf.getBoolean(
+      HConstants.HBASE_REGIONSERVER_BULKLOAD_INDEPENDENCE_QUEUE_ENABLED,
+      HConstants.DEFAULT_HBASE_REGIONSERVER_BULKLOAD_INDEPENDENCE_QUEUE_ENABLED);
+    if (bulkLoadIndependenceQueueEnabled) {
+      return new SimpleRpcScheduler(conf, handlerCount,
+        conf.getInt(HConstants.REGION_SERVER_HIGH_PRIORITY_HANDLER_COUNT,
+          HConstants.DEFAULT_REGION_SERVER_HIGH_PRIORITY_HANDLER_COUNT),
+        conf.getInt(HConstants.REGION_SERVER_REPLICATION_HANDLER_COUNT,
+          HConstants.DEFAULT_REGION_SERVER_REPLICATION_HANDLER_COUNT),
+        conf.getInt(HConstants.MASTER_META_TRANSITION_HANDLER_COUNT,
+          HConstants.MASTER__META_TRANSITION_HANDLER_COUNT_DEFAULT),
+        conf.getInt(HConstants.REGION_SERVER_BULKLOAD_HANDLER_COUNT,
+          HConstants.DEFAULT_REGION_SERVER_BULKLOAD_HANDLER_COUNT),
+        priority, server, HConstants.QOS_THRESHOLD);
+    } else {
+      return new SimpleRpcScheduler(conf, handlerCount,
+        conf.getInt(HConstants.REGION_SERVER_HIGH_PRIORITY_HANDLER_COUNT,
+          HConstants.DEFAULT_REGION_SERVER_HIGH_PRIORITY_HANDLER_COUNT),
+        conf.getInt(HConstants.REGION_SERVER_REPLICATION_HANDLER_COUNT,
+          HConstants.DEFAULT_REGION_SERVER_REPLICATION_HANDLER_COUNT),
+        conf.getInt(HConstants.MASTER_META_TRANSITION_HANDLER_COUNT,
+          HConstants.MASTER__META_TRANSITION_HANDLER_COUNT_DEFAULT),
+        priority, server, HConstants.QOS_THRESHOLD);
+    }
   }
 }