@Override
  protected CompletionStage<Collection<M>> handleMessageAsync(M message, MessageCollector collector,
      TaskCoordinator coordinator) {
    Object key, value;
    if (outputStream.isKeyed()) {
      key = ((KV) message).getKey();
      value = ((KV) message).getValue();
    } else {
      key = null;
      value = message;
    }

    collector.send(new OutgoingMessageEnvelope(systemStream, null, key, value));
    return CompletableFuture.completedFuture(Collections.singleton(message));
  }
An operator that sends incoming messages to an output {@link SystemStream}.
samza-core/src/main/java/org/apache/samza/operators/impl/OutputOperatorImpl.java