@Override
  protected CompletionStage<Collection<Object>> handleMessageAsync(IncomingMessageEnvelope message,
      MessageCollector collector, TaskCoordinator coordinator) {
    Object result;
    InputTransformer transformer = inputOpSpec.getTransformer();
    if (transformer != null) {
      result = transformer.apply(message);
    } else {
      result = this.inputOpSpec.isKeyed() ? KV.of(message.getKey(), message.getMessage()) : message.getMessage();
    }

    Collection<Object> output = Optional.ofNullable(result)
        .map(Collections::singletonList)
        .orElse(Collections.emptyList());

    return CompletableFuture.completedFuture(output);
  }
An operator that builds the input message from the incoming message.
samza-core/src/main/java/org/apache/samza/operators/impl/InputOperatorImpl.java