protected void handleGetAsyncBinaryDocument(BinaryDocument binaryDocument, CompletableFuture<V> future, String key) {
    ByteBuf buffer = binaryDocument.content();
    try {
      byte[] bytes;
      if (buffer.hasArray() && buffer.arrayOffset() == 0 && buffer.readableBytes() == buffer.array().length) {
        bytes = buffer.array();
      } else {
        bytes = new byte[buffer.readableBytes()];
        buffer.readBytes(bytes);
      }
      future.complete(valueSerde.fromBytes(bytes));
    } catch (Exception e) {
      future.completeExceptionally(
          new SamzaException(String.format("Failed to deserialize value of key %s with given serde", key), e));
    } finally {
      ReferenceCountUtil.release(buffer);
    }
  }
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