@Override
  public CompletableFuture<V> getAsync(String key) {
    Preconditions.checkArgument(StringUtils.isNotBlank(key), "key must not be null, empty or blank");
    CompletableFuture<V> future = new CompletableFuture<>();
    Single<? extends Document<?>> singleObservable =
        bucket.async().get(key, documentType, timeout.toMillis(), TimeUnit.MILLISECONDS).toSingle();
    singleObservable.subscribe(new SingleSubscriber<Document<?>>() {
      @Override
      public void onSuccess(Document<?> document) {
        if (document != null) {
          if (document instanceof BinaryDocument) {
            handleGetAsyncBinaryDocument((BinaryDocument) document, future, key);
          } else {
            // V is of type JsonObject
            future.complete((V) document.content());
          }
        } else {
          // The Couchbase async client should not return null
          future.completeExceptionally(new SamzaException(String.format("Got unexpected null value from key %s", key)));
        }
      }

      @Override
      public void onError(Throwable throwable) {
        if (throwable instanceof NoSuchElementException) {
          // There is no element returned by the observable, meaning the key doesn't exist.
          future.complete(null);
        } else {
          future.completeExceptionally(new SamzaException(String.format("Failed to get key %s", key), throwable));
        }
      }
    });
    return future;
  }
Construct an instance of {@link CouchbaseTableReadFunction}.
@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/CouchbaseTableReadFunction.java