@Override
  public void mutate(List<? extends Mutation> mutations) throws IOException {
    List<CompletableFuture<Void>> fs = mutator.mutate(mutations);
    for (int i = 0, n = fs.size(); i < n; i++) {
      CompletableFuture<Void> toComplete = new CompletableFuture<>();
      futures.add(toComplete);
      Mutation mutation = mutations.get(i);
      long heapSize = mutation.heapSize();
      bufferedSize.addAndGet(heapSize);
      addListener(fs.get(i), (r, e) -> {
        synchronized (this) {
          futures.remove(toComplete);
          bufferedSize.addAndGet(-heapSize);
          if (e != null) {
            errors.add(Pair.newPair(mutation, e));
            toComplete.completeExceptionally(e);
          } else {
            toComplete.complete(r);
          }
        }
      });
    }
    synchronized (this) {
      if (bufferedSize.get() > mutator.getWriteBufferSize() * 2) {
        // We have too many mutations which are not completed yet, let's call a flush to release the
        // memory to prevent OOM
        // We use buffer size * 2 is because that, the async buffered mutator will flush
        // automatically when the write buffer size limit is reached, so usually we do not need to
        // call flush explicitly if the buffered size is only a little larger than the buffer size
        // limit. But if the buffered size is too large(2 times of the buffer size), we still need
        // to block here to prevent OOM.
        internalFlush();
      } else if (!errors.isEmpty()) {
        RetriesExhaustedWithDetailsException error = makeError();
        listener.onException(error, this);
      }
    }
  }
{@link BufferedMutator} implementation based on {@link AsyncBufferedMutator}.
hbase-client/src/main/java/org/apache/hadoop/hbase/client/BufferedMutatorOverAsyncBufferedMutator.java