@Override
  protected CompletionStage<Collection<Void>> handleMessageAsync(M message, MessageCollector collector,
      TaskCoordinator coordinator) {
    K key = keyFunction.apply(message);
    V value = valueFunction.apply(message);
    Long partitionKey = key == null ? 0L : null;
    collector.send(new OutgoingMessageEnvelope(systemStream, partitionKey, key, value));
    return CompletableFuture.completedFuture(Collections.emptyList());
  }
An operator that sends sends messages to an output {@link SystemStream} for repartitioning them.
samza-core/src/main/java/org/apache/samza/operators/impl/PartitionByOperatorImpl.java