@Override
  protected CompletionStage<Collection<KV<K, UpdateMessage<U, V>>>> handleMessageAsync(KV<K, UpdateMessage<U, V>> message,
      MessageCollector collector, TaskCoordinator coordinator) {
    final UpdateOptions options = spec.getUpdateOptions();
    final CompletableFuture<Void> updateFuture = table.updateAsync(message.getKey(), message.getValue().getUpdate());

    return updateFuture
        .handle((result, ex) -> {
          if (ex == null) {
            // success, no need to Put a default value
            return false;
          } else if (ex.getCause() instanceof RecordNotFoundException && message.getValue().hasDefault()) {
            // If an update fails for a given key due to a RecordDoesNotExistException exception thrown and
            // a default is provided and the UpdateOptions is set to UPDATE_WITH_DEFAULTS, then attempt
            // to PUT a default record for the key and then apply the update.
            if (options == UpdateOptions.UPDATE_WITH_DEFAULTS) {
              return true;
            } else {
              throw new SamzaException("Put default failed for update as the Update options was set to " + options +
                  ". Please use UpdateOptions.UPDATE_WITH_DEFAULTS instead.");
            }
          } else {
            throw new SamzaException("Update failed with exception: ", ex);
          }
        })
        .thenCompose(shouldPutDefault -> {
          if (shouldPutDefault) {
            final CompletableFuture<Void> putFuture = table.putAsync(message.getKey(), message.getValue().getDefault());
            return putFuture
                .exceptionally(ex -> {
                  LOG.warn("PUT default failed due to an exception. Ignoring the exception and proceeding with update. "
                          + "The exception encountered is: ", ex);
                  return null;
                })
                .thenCompose(res -> table.updateAsync(message.getKey(), message.getValue().getUpdate()))
                .exceptionally(ex -> {
                  throw new SamzaException("Update after Put default failed with exception: ", ex);
                });
          } else {
            return CompletableFuture.completedFuture(null);
          }
        }).thenApply(result -> Collections.singleton(message));
  }
Implementation of a send-to-table operator that applies updates to existing records
in the table. If there is no pre-existing record, based on Table's implementation it might attempt to write a
default record and then applies an update.

@param <K> the type of the record key
@param <V> the type of the record value
@param <U> the type of the update applied to this table
samza-core/src/main/java/org/apache/samza/operators/impl/SendToTableWithUpdateOperatorImpl.java