public final CompletionStage<Void> onMessageAsync(M message, MessageCollector collector,
      TaskCoordinator coordinator) {
    this.numMessage.inc();
    long startNs = this.highResClock.nanoTime();
    CompletionStage<Collection<RM>> completableResultsFuture;
    try {
      completableResultsFuture = handleMessageAsync(message, collector, coordinator);
    } catch (ClassCastException e) {
      String actualType = e.getMessage().replaceFirst(" cannot be cast to .*", "");
      String expectedType = e.getMessage().replaceFirst(".* cannot be cast to ", "");
      throw new SamzaException(
          String.format("Error applying operator %s (created at %s) to its input message. "
                  + "Expected input message to be of type %s, but found it to be of type %s. "
                  + "Are Serdes for the inputs to this operator configured correctly?",
              getOpImplId(), getOperatorSpec().getSourceLocation(), expectedType, actualType), e);
    }

    CompletionStage<Void> result = composeFutureWithExecutor(completableResultsFuture, results -> {
      long endNs = this.highResClock.nanoTime();
      this.handleMessageNs.update(endNs - startNs);

      return CompletableFuture.allOf(results.stream()
          .flatMap(r -> this.registeredOperators.stream().map(op -> op.onMessageAsync(r, collector, coordinator)))
          .toArray(CompletableFuture[]::new));
    });

    WatermarkFunction watermarkFn = getOperatorSpec().getWatermarkFn();
    if (watermarkFn != null) {
      // check whether there is new watermark emitted from the user function
      Long outputWm = watermarkFn.getOutputWatermark();
      return composeFutureWithExecutor(result, ignored -> propagateWatermark(outputWm, collector, coordinator));
    }

    return result;
  }
Register an operator that this operator should propagate its results to.

@param nextOperator  the next operator to propagate results to
samza-core/src/main/java/org/apache/samza/operators/impl/OperatorImpl.java