code: private ApiFuture<?> offer(Mutation mutation) {
      ApiFuture<?> future = null;
      try {
        if (mutation == null) {
          future = ApiFutures.immediateFailedFuture(
            new IllegalArgumentException("Cannot perform a mutation on a null object."));
        } else if (mutation instanceof Put) {
          future = bulkMutation.add(adapter.adaptEntry((Put) mutation));
        } else if (mutation instanceof Delete) {
          future = bulkMutation.add(adapter.adaptEntry((Delete) mutation));
        } else if (mutation instanceof Increment) {
          future = dataClient.readModifyWriteRowAsync(adapter.adapt((Increment) mutation));
          operationAccountant.registerOperation(ApiFutureUtil.adapt(future));
        } else if (mutation instanceof Append) {
          future = dataClient.readModifyWriteRowAsync(adapter.adapt((Append) mutation));
          operationAccountant.registerOperation(ApiFutureUtil.adapt(future));
        } else {
          future = ApiFutures.immediateFailedFuture(new IllegalArgumentException(
              "Encountered unknown mutation type: " + mutation.getClass()));
        }
      } catch (Exception e) {
        // issueRequest(mutation) could throw an Exception for validation issues. Remove the heapsize
        // and inflight rpc count.
        future = ApiFutures.immediateFailedFuture(e);
      }
      return future;
    }
docs: Send the operations to the async executor asynchronously.  The conversion from hbase
  object to cloud bigtable proto and the async call both take time (microseconds worth) that
  could be parallelized, or at least removed from the user's thread.