private CompletionStage<Void> onEndOfStream(MessageCollector collector, TaskCoordinator coordinator) {
    CompletionStage<Void> endOfStreamFuture = CompletableFuture.completedFuture(null);

    if (inputStreams.stream().allMatch(input -> eosStates.isEndOfStream(input))) {
      Collection<RM> results = handleEndOfStream(collector, coordinator);

      CompletionStage<Void> resultFuture = CompletableFuture.allOf(
          results.stream()
              .flatMap(r -> this.registeredOperators.stream()
                  .map(op -> op.onMessageAsync(r, collector, coordinator)))
              .toArray(CompletableFuture[]::new));

      endOfStreamFuture = composeFutureWithExecutor(resultFuture, x -> CompletableFuture.allOf(
          this.registeredOperators.stream()
              .map(op -> op.onEndOfStream(collector, coordinator))
              .toArray(CompletableFuture[]::new)));
    }

    return endOfStreamFuture;
  }
Invoke handleEndOfStream() if all the input streams to the current operator reach the end.
Propagate the end-of-stream to downstream operators.
@param collector message collector
@param coordinator task coordinator
samza-core/src/main/java/org/apache/samza/operators/impl/OperatorImpl.java