public final CompletionStage<Void> aggregateEndOfStream(EndOfStreamMessage eos, SystemStreamPartition ssp, MessageCollector collector,
      TaskCoordinator coordinator) {
    LOG.info("Received end-of-stream message from task {} in {}", eos.getTaskName(), ssp);
    eosStates.update(eos, ssp);

    SystemStream stream = ssp.getSystemStream();
    CompletionStage<Void> endOfStreamFuture = CompletableFuture.completedFuture(null);

    if (eosStates.isEndOfStream(stream)) {
      LOG.info("Input {} reaches the end for task {}", stream.toString(), taskName.getTaskName());
      if (eos.getTaskName() != null && shouldTaskBroadcastToOtherPartitions(ssp)) {
        // This is the aggregation task, which already received all the eos messages from upstream
        // broadcast the end-of-stream to all the peer partitions
        // additionally if elasiticty is enabled
        // then only one of the elastic tasks of the ssp will broadcast
        controlMessageSender.broadcastToOtherPartitions(new EndOfStreamMessage(), ssp, collector);
      }

      // populate the end-of-stream through the dag
      endOfStreamFuture = acceptFutureWithExecutor(onEndOfStream(collector, coordinator), result -> {
        if (eosStates.allEndOfStream()) {
          // all inputs have been end-of-stream, shut down the task
          LOG.info("All input streams have reached the end for task {}", taskName.getTaskName());
          coordinator.commit(TaskCoordinator.RequestScope.CURRENT_TASK);
          coordinator.shutdown(TaskCoordinator.RequestScope.CURRENT_TASK);
        }
      });
    }

    return endOfStreamFuture;
  }
Aggregate {@link EndOfStreamMessage} from each ssp of the stream.
Invoke onEndOfStream() if the stream reaches the end.
@param eos {@link EndOfStreamMessage} object
@param ssp system stream partition
@param collector message collector
@param coordinator task coordinator
samza-core/src/main/java/org/apache/samza/operators/impl/OperatorImpl.java