private void batch(TableName tableName, Collection<List<Row>> allRows, int batchRowSizeThreshold)
    throws IOException {
    if (allRows.isEmpty()) {
      return;
    }
    AsyncTable<?> table = getConnection().getTable(tableName);
    List<Future<?>> futures = new ArrayList<>();
    for (List<Row> rows : allRows) {
      List<List<Row>> batchRows;
      if (rows.size() > batchRowSizeThreshold) {
        batchRows = Lists.partition(rows, batchRowSizeThreshold);
      } else {
        batchRows = Collections.singletonList(rows);
      }
      futures.addAll(batchRows.stream().map(table::batchAll).collect(Collectors.toList()));
    }
    // Here we will always wait until all futures are finished, even if there are failures when
    // getting from a future in the middle. This is because this method may be called in a rpc call,
    // so the batch operations may reference some off heap cells(through CellScanner). If we return
    // earlier here, the rpc call may be finished and they will release the off heap cells before
    // some of the batch operations finish, and then cause corrupt data or even crash the region
    // server. See HBASE-28584 and HBASE-28850 for more details.
    IOException error = null;
    for (Future<?> future : futures) {
      try {
        FutureUtils.get(future);
      } catch (RetriesExhaustedException e) {
        IOException ioe;
        if (e.getCause() instanceof TableNotFoundException) {
          ioe = new TableNotFoundException("'" + tableName + "'");
        } else {
          ioe = e;
        }
        if (error == null) {
          error = ioe;
        } else {
          error.addSuppressed(ioe);
        }
      }
    }
    if (error != null) {
      throw error;
    }
  }
Do the changes and handle the pool
@param tableName             table to insert into
@param allRows               list of actions
@param batchRowSizeThreshold rowSize threshold for batch mutation
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java