Experiment related to apache/fluo#978. This very incomplete and may be a terrible design. This is the result of an initial experiment just to think about the high level structure.

AsyncCommit.java

  private ExecutorService syncExecService = null;

  // writes the conditional mutation that attempts to lock the primary column.
  CompletableFuture<Boolean> lockPrimary() {
    Flutation lockPrimaryCondMut = null;
    CompletableFuture<Status> future =
        asyncCW.write(lockPrimaryCondMut).thenApply(Result::getStatus);

    return future.thenCompose(status -> {
      if (status == Status.UNKNOWN) {
        // following supplier should try to resolve UKNOWN status in a while loop
        Supplier<Status> syncStatusResolver = null;
        return CompletableFuture.supplyAsync(syncStatusResolver, syncExecService)
            .thenApply(s -> s == Status.ACCEPTED);
      } else {
        return CompletableFuture.completedFuture(status == Status.ACCEPTED);
      }
    });
  };


  // attempts to lock all other columns
  CompletableFuture<Boolean> lockOther() {
    return null;
  }

  // attempts to commit primary column
  CompletableFuture<Boolean> commitPrimary() {
    return null;
  }

  // cleans up locks for all other columns
  CompletableFuture<Void> finishCommit() {
    return null;
  }

  CompletableFuture<Void> handleFailedPrimaryLock() {
    // TODO check for ack and read unread in sync pool
    Supplier<Void> failAction = null;
    return CompletableFuture.supplyAsync(failAction, syncExecService);
  }

  CompletableFuture<Void> handleFailedOtherLocks() {
    Supplier<Void> failAction = null;
    return CompletableFuture.supplyAsync(failAction, syncExecService);
  }

  CompletableFuture<Void> handleFailedPrimaryCommit() {
    Supplier<Void> failAction = null;
    return CompletableFuture.supplyAsync(failAction, syncExecService);
  }

  CompletableFuture<Boolean> commit() {
    return lockPrimary().thenCompose(primaryLocked -> {
      if (primaryLocked) {
        return lockOther().thenCompose(othersLocked -> {
          if (othersLocked) {
            return commitPrimary().thenCompose(primaryCommitted -> {
              if (primaryCommitted) {
                return finishCommit().thenApply(v -> true);
              } else {
                return handleFailedPrimaryCommit().thenApply(v -> false);
              }
            });
          } else {
            return handleFailedOtherLocks().thenApply(v -> false);
          }
        });
      } else {
        return handleFailedPrimaryLock().thenApply(v -> false);
      }
    });
  }