@Override
  public MultiResponse multi(final RpcController rpcc, final MultiRequest request)
    throws ServiceException {
    try {
      checkOpen();
    } catch (IOException ie) {
      throw new ServiceException(ie);
    }

    checkBatchSizeAndLogLargeSize(request);

    // rpc controller is how we bring in data via the back door; it is unprotobuf'ed data.
    // It is also the conduit via which we pass back data.
    HBaseRpcController controller = (HBaseRpcController) rpcc;
    CellScanner cellScanner = controller != null ? getAndReset(controller) : null;

    long nonceGroup = request.hasNonceGroup() ? request.getNonceGroup() : HConstants.NO_NONCE;

    MultiResponse.Builder responseBuilder = MultiResponse.newBuilder();
    RegionActionResult.Builder regionActionResultBuilder = RegionActionResult.newBuilder();
    this.rpcMultiRequestCount.increment();
    this.requestCount.increment();
    ActivePolicyEnforcement spaceQuotaEnforcement = getSpaceQuotaManager().getActiveEnforcements();

    // We no longer use MultiRequest#condition. Instead, we use RegionAction#condition. The
    // following logic is for backward compatibility as old clients still use
    // MultiRequest#condition in case of checkAndMutate with RowMutations.
    if (request.hasCondition()) {
      if (request.getRegionActionList().isEmpty()) {
        // If the region action list is empty, do nothing.
        responseBuilder.setProcessed(true);
        return responseBuilder.build();
      }

      RegionAction regionAction = request.getRegionAction(0);

      // When request.hasCondition() is true, regionAction.getAtomic() should be always true. So
      // we can assume regionAction.getAtomic() is true here.
      assert regionAction.getAtomic();

      OperationQuota quota;
      HRegion region;
      RegionSpecifier regionSpecifier = regionAction.getRegion();

      try {
        region = getRegion(regionSpecifier);
        quota = getRpcQuotaManager().checkBatchQuota(region, regionAction.getActionList(),
          regionAction.hasCondition());
      } catch (IOException e) {
        failRegionAction(responseBuilder, regionActionResultBuilder, regionAction, cellScanner, e);
        return responseBuilder.build();
      }

      try {
        boolean rejectIfFromClient = shouldRejectRequestsFromClient(region);
        // We only allow replication in standby state and it will not set the atomic flag.
        if (rejectIfFromClient) {
          failRegionAction(responseBuilder, regionActionResultBuilder, regionAction, cellScanner,
            new DoNotRetryIOException(
              region.getRegionInfo().getRegionNameAsString() + " is in STANDBY state"));
          return responseBuilder.build();
        }

        try {
          CheckAndMutateResult result = checkAndMutate(region, regionAction.getActionList(),
            cellScanner, request.getCondition(), nonceGroup, spaceQuotaEnforcement);
          responseBuilder.setProcessed(result.isSuccess());
          ClientProtos.ResultOrException.Builder resultOrExceptionOrBuilder =
            ClientProtos.ResultOrException.newBuilder();
          for (int i = 0; i < regionAction.getActionCount(); i++) {
            // To unify the response format with doNonAtomicRegionMutation and read through
            // client's AsyncProcess we have to add an empty result instance per operation
            resultOrExceptionOrBuilder.clear();
            resultOrExceptionOrBuilder.setIndex(i);
            regionActionResultBuilder.addResultOrException(resultOrExceptionOrBuilder.build());
          }
        } catch (IOException e) {
          rpcServer.getMetrics().exception(e);
          // As it's an atomic operation with a condition, we may expect it's a global failure.
          regionActionResultBuilder.setException(ResponseConverter.buildException(e));
        }
      } finally {
        quota.close();
      }

      responseBuilder.addRegionActionResult(regionActionResultBuilder.build());
      ClientProtos.RegionLoadStats regionLoadStats = region.getLoadStatistics();
      if (regionLoadStats != null) {
        responseBuilder.setRegionStatistics(MultiRegionLoadStats.newBuilder()
          .addRegion(regionSpecifier).addStat(regionLoadStats).build());
      }
      return responseBuilder.build();
    }

    // this will contain all the cells that we need to return. It's created later, if needed.
    List<ExtendedCellScannable> cellsToReturn = null;
    RegionScannersCloseCallBack closeCallBack = null;
    RpcCallContext context = RpcServer.getCurrentCall().orElse(null);
    Map<RegionSpecifier, ClientProtos.RegionLoadStats> regionStats =
      new HashMap<>(request.getRegionActionCount());

    for (RegionAction regionAction : request.getRegionActionList()) {
      OperationQuota quota;
      HRegion region;
      RegionSpecifier regionSpecifier = regionAction.getRegion();
      regionActionResultBuilder.clear();

      try {
        region = getRegion(regionSpecifier);
        quota = getRpcQuotaManager().checkBatchQuota(region, regionAction.getActionList(),
          regionAction.hasCondition());
      } catch (IOException e) {
        failRegionAction(responseBuilder, regionActionResultBuilder, regionAction, cellScanner, e);
        continue; // For this region it's a failure.
      }

      try {
        boolean rejectIfFromClient = shouldRejectRequestsFromClient(region);

        if (regionAction.hasCondition()) {
          // We only allow replication in standby state and it will not set the atomic flag.
          if (rejectIfFromClient) {
            failRegionAction(responseBuilder, regionActionResultBuilder, regionAction, cellScanner,
              new DoNotRetryIOException(
                region.getRegionInfo().getRegionNameAsString() + " is in STANDBY state"));
            continue;
          }

          try {
            ClientProtos.ResultOrException.Builder resultOrExceptionOrBuilder =
              ClientProtos.ResultOrException.newBuilder();
            if (regionAction.getActionCount() == 1) {
              CheckAndMutateResult result =
                checkAndMutate(region, quota, regionAction.getAction(0).getMutation(), cellScanner,
                  regionAction.getCondition(), nonceGroup, spaceQuotaEnforcement, context);
              regionActionResultBuilder.setProcessed(result.isSuccess());
              resultOrExceptionOrBuilder.setIndex(0);
              if (result.getResult() != null) {
                resultOrExceptionOrBuilder.setResult(ProtobufUtil.toResult(result.getResult()));
              }

              if (result.getMetrics() != null) {
                resultOrExceptionOrBuilder
                  .setMetrics(ProtobufUtil.toQueryMetrics(result.getMetrics()));
              }

              regionActionResultBuilder.addResultOrException(resultOrExceptionOrBuilder.build());
            } else {
              CheckAndMutateResult result = checkAndMutate(region, regionAction.getActionList(),
                cellScanner, regionAction.getCondition(), nonceGroup, spaceQuotaEnforcement);
              regionActionResultBuilder.setProcessed(result.isSuccess());
              for (int i = 0; i < regionAction.getActionCount(); i++) {
                if (i == 0 && result.getResult() != null) {
                  // Set the result of the Increment/Append operations to the first element of the
                  // ResultOrException list
                  resultOrExceptionOrBuilder.setIndex(i);
                  regionActionResultBuilder.addResultOrException(resultOrExceptionOrBuilder
                    .setResult(ProtobufUtil.toResult(result.getResult())).build());
                  continue;
                }
                // To unify the response format with doNonAtomicRegionMutation and read through
                // client's AsyncProcess we have to add an empty result instance per operation
                resultOrExceptionOrBuilder.clear();
                resultOrExceptionOrBuilder.setIndex(i);
                regionActionResultBuilder.addResultOrException(resultOrExceptionOrBuilder.build());
              }
            }
          } catch (IOException e) {
            rpcServer.getMetrics().exception(e);
            // As it's an atomic operation with a condition, we may expect it's a global failure.
            regionActionResultBuilder.setException(ResponseConverter.buildException(e));
          }
        } else if (regionAction.hasAtomic() && regionAction.getAtomic()) {
          // We only allow replication in standby state and it will not set the atomic flag.
          if (rejectIfFromClient) {
            failRegionAction(responseBuilder, regionActionResultBuilder, regionAction, cellScanner,
              new DoNotRetryIOException(
                region.getRegionInfo().getRegionNameAsString() + " is in STANDBY state"));
            continue;
          }
          try {
            doAtomicBatchOp(regionActionResultBuilder, region, quota, regionAction.getActionList(),
              cellScanner, nonceGroup, spaceQuotaEnforcement);
            regionActionResultBuilder.setProcessed(true);
            // We no longer use MultiResponse#processed. Instead, we use
            // RegionActionResult#processed. This is for backward compatibility for old clients.
            responseBuilder.setProcessed(true);
          } catch (IOException e) {
            rpcServer.getMetrics().exception(e);
            // As it's atomic, we may expect it's a global failure.
            regionActionResultBuilder.setException(ResponseConverter.buildException(e));
          }
        } else {
          if (
            rejectIfFromClient && regionAction.getActionCount() > 0
              && !isReplicationRequest(regionAction.getAction(0))
          ) {
            // fail if it is not a replication request
            failRegionAction(responseBuilder, regionActionResultBuilder, regionAction, cellScanner,
              new DoNotRetryIOException(
                region.getRegionInfo().getRegionNameAsString() + " is in STANDBY state"));
            continue;
          }
          // doNonAtomicRegionMutation manages the exception internally
          if (context != null && closeCallBack == null) {
            // An RpcCallBack that creates a list of scanners that needs to perform callBack
            // operation on completion of multiGets.
            // Set this only once
            closeCallBack = new RegionScannersCloseCallBack();
            context.setCallBack(closeCallBack);
          }
          cellsToReturn = doNonAtomicRegionMutation(region, quota, regionAction, cellScanner,
            regionActionResultBuilder, cellsToReturn, nonceGroup, closeCallBack, context,
            spaceQuotaEnforcement);
        }
      } finally {
        quota.close();
      }

      responseBuilder.addRegionActionResult(regionActionResultBuilder.build());
      ClientProtos.RegionLoadStats regionLoadStats = region.getLoadStatistics();
      if (regionLoadStats != null) {
        regionStats.put(regionSpecifier, regionLoadStats);
      }
    }
    // Load the controller with the Cells to return.
    if (cellsToReturn != null && !cellsToReturn.isEmpty() && controller != null) {
      controller.setCellScanner(PrivateCellUtil.createExtendedCellScanner(cellsToReturn));
    }

    MultiRegionLoadStats.Builder builder = MultiRegionLoadStats.newBuilder();
    for (Entry<RegionSpecifier, ClientProtos.RegionLoadStats> stat : regionStats.entrySet()) {
      builder.addRegion(stat.getKey());
      builder.addStat(stat.getValue());
    }
    responseBuilder.setRegionStatistics(builder);
    return responseBuilder.build();
  }
Execute multiple actions on a table: get, mutate, and/or execCoprocessor
@param rpcc    the RPC controller
@param request the multi request
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java