@Override
  public <K, V, U> MessageStream<KV<K, UpdateMessage<U, V>>> sendTo(Table<KV<K, V>> table, UpdateOptions updateOptions) {
    String opId = this.streamAppDesc.getNextOpId(OpCode.SEND_TO_WITH_UPDATE);
    SendToTableWithUpdateOperatorSpec<K, V, U> op =
        OperatorSpecs.createSendToTableWithUpdateOperatorSpec(((TableImpl) table).getTableId(), opId, updateOptions);
    this.operatorSpec.registerNextOperatorSpec(op);
    return new MessageStreamImpl<>(this.streamAppDesc, op);
  }
The {@link OperatorSpec} associated with this {@link MessageStreamImpl}
samza-core/src/main/java/org/apache/samza/operators/MessageStreamImpl.java