private CompletionStage<Void> onDrainOfStream(MessageCollector collector, TaskCoordinator coordinator) {
    CompletionStage<Void> drainFuture = CompletableFuture.completedFuture(null);
    if (inputStreams.stream().allMatch(input -> drainStates.isDrained(input))) {
      Collection<RM> results = handleDrain(collector, coordinator);
      CompletionStage<Void> resultFuture = CompletableFuture.allOf(
          results.stream()
              .flatMap(r -> this.registeredOperators.stream()
                  .map(op -> op.onMessageAsync(r, collector, coordinator)))
              .toArray(CompletableFuture[]::new));

      // propagate DrainMessage to downstream operators
      drainFuture = composeFutureWithExecutor(resultFuture, x -> CompletableFuture.allOf(
          this.registeredOperators.stream()
              .map(op -> op.onDrainOfStream(collector, coordinator))
              .toArray(CompletableFuture[]::new)));
    }

    return drainFuture;
  }
Invoke {@link #handleDrain(MessageCollector, TaskCoordinator)} if all the input streams to the current operator
have encountered drain message.
Propagate the drain to downstream operators.
@param collector message collector
@param coordinator task coordinator
samza-core/src/main/java/org/apache/samza/operators/impl/OperatorImpl.java