@Override
  protected CompletionStage<Collection<KV<K, V>>> handleMessageAsync(KV<K, V> message, MessageCollector collector,
      TaskCoordinator coordinator) {
    if (message.getValue() instanceof UpdateMessage) {
      throw new SamzaException("Incorrect use of .sendTo operator with UpdateMessage value type. "
          + "Please use the following method on MessageStream- sendTo(Table<KV<K, UpdateMessage<U, V>>> table,"
          + "UpdateOptions updateOptions).");
    }
    return table.putAsync(message.getKey(), message.getValue())
        .thenApply(result -> Collections.singleton(message));
  }
Implementation of a send-stream-to-table operator that stores the record
in the table.

@param <K> the type of the record key
@param <V> the type of the record value
samza-core/src/main/java/org/apache/samza/operators/impl/SendToTableOperatorImpl.java