protected <T> CompletableFuture<T>  asyncWriteHelper(Observable<? extends Document> observable, String errorMessage,
      boolean isVoid) {
    CompletableFuture<T> future = new CompletableFuture<>();
    observable.toSingle().subscribe(new SingleSubscriber<Document>() {
      @Override
      public void onSuccess(Document document) {
        if (isVoid) {
          future.complete(null);
        } else {
          future.complete((T) document.content());
        }
      }

      @Override
      public void onError(Throwable error) {
        future.completeExceptionally(new SamzaException(errorMessage, error));
      }
    });
    return future;
  }
Construct an instance of {@link CouchbaseTableWriteFunction}.
@param bucketName Name of the couchbase bucket
@param clusterNodes Some Hosts of the Couchbase cluster. Recommended to provide more than one nodes so that if
                    the first node could not be connected, other nodes can be tried.
@param valueClass Type of values
samza-kv-couchbase/src/main/java/org/apache/samza/table/remote/couchbase/CouchbaseTableWriteFunction.java