@Override
  public CompletableFuture<Void> putAsync(String key, V record) {
    Preconditions.checkArgument(StringUtils.isNotBlank(key), "key must not be null, empty or blank");
    Preconditions.checkArgument(!key.contains(" "), String.format("key should not contain spaces: %s", key));
    Preconditions.checkNotNull(record);
    Document<?> document = record instanceof JsonObject
        ? JsonDocument.create(key, (int) ttl.getSeconds(), (JsonObject) record)
        : BinaryDocument.create(key, (int) ttl.getSeconds(), Unpooled.copiedBuffer(valueSerde.toBytes(record)));
    return asyncWriteHelper(
        bucket.async().upsert(document, timeout.toMillis(), TimeUnit.MILLISECONDS),
        String.format("Failed to insert key %s into bucket %s", key, bucketName));
  }
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