@Override
  public BulkLoadHFileResponse bulkLoadHFile(final RpcController controller,
    final BulkLoadHFileRequest request) throws ServiceException {
    long start = EnvironmentEdgeManager.currentTime();
    List<String> clusterIds = new ArrayList<>(request.getClusterIdsList());
    if (clusterIds.contains(this.server.getClusterId())) {
      return BulkLoadHFileResponse.newBuilder().setLoaded(true).build();
    } else {
      clusterIds.add(this.server.getClusterId());
    }
    try {
      checkOpen();
      requestCount.increment();
      HRegion region = getRegion(request.getRegion());
      final boolean spaceQuotaEnabled = QuotaUtil.isQuotaEnabled(getConfiguration());
      long sizeToBeLoaded = -1;

      // Check to see if this bulk load would exceed the space quota for this table
      if (spaceQuotaEnabled) {
        ActivePolicyEnforcement activeSpaceQuotas = getSpaceQuotaManager().getActiveEnforcements();
        SpaceViolationPolicyEnforcement enforcement =
          activeSpaceQuotas.getPolicyEnforcement(region);
        if (enforcement != null) {
          // Bulk loads must still be atomic. We must enact all or none.
          List<String> filePaths = new ArrayList<>(request.getFamilyPathCount());
          for (FamilyPath familyPath : request.getFamilyPathList()) {
            filePaths.add(familyPath.getPath());
          }
          // Check if the batch of files exceeds the current quota
          sizeToBeLoaded = enforcement.computeBulkLoadSize(getFileSystem(filePaths), filePaths);
        }
      }
      // secure bulk load
      Map<byte[], List<Path>> map =
        server.getSecureBulkLoadManager().secureBulkLoadHFiles(region, request, clusterIds);
      BulkLoadHFileResponse.Builder builder = BulkLoadHFileResponse.newBuilder();
      builder.setLoaded(map != null);
      if (map != null) {
        // Treat any negative size as a flag to "ignore" updating the region size as that is
        // not possible to occur in real life (cannot bulk load a file with negative size)
        if (spaceQuotaEnabled && sizeToBeLoaded > 0) {
          if (LOG.isTraceEnabled()) {
            LOG.trace("Incrementing space use of " + region.getRegionInfo() + " by "
              + sizeToBeLoaded + " bytes");
          }
          // Inform space quotas of the new files for this region
          getSpaceQuotaManager().getRegionSizeStore().incrementRegionSize(region.getRegionInfo(),
            sizeToBeLoaded);
        }
      }
      return builder.build();
    } catch (IOException ie) {
      throw new ServiceException(ie);
    } finally {
      final MetricsRegionServer metricsRegionServer = server.getMetrics();
      if (metricsRegionServer != null) {
        metricsRegionServer.updateBulkLoad(EnvironmentEdgeManager.currentTime() - start);
      }
    }
  }
Atomically bulk load several HFiles into an open region
@return true if successful, false is failed but recoverably (no action)
@throws ServiceException if failed unrecoverably
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java