public final CompletionStage<Void> aggregateWatermark(WatermarkMessage watermarkMessage, SystemStreamPartition ssp,
      MessageCollector collector, TaskCoordinator coordinator) {
    LOG.debug("Received watermark {} from {}", watermarkMessage.getTimestamp(), ssp);
    watermarkStates.update(watermarkMessage, ssp);
    long watermark = watermarkStates.getWatermark(ssp.getSystemStream());
    CompletionStage<Void> watermarkFuture = CompletableFuture.completedFuture(null);

    if (currentWatermark < watermark) {
      LOG.debug("Got watermark {} from stream {}", watermark, ssp.getSystemStream());

      if (watermarkMessage.getTaskName() != null && shouldTaskBroadcastToOtherPartitions(ssp)) {
        // This is the aggregation task, which already received all the watermark messages from upstream
        // broadcast the watermark 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 WatermarkMessage(watermark), ssp, collector);
      }
      // populate the watermark through the dag
      watermarkFuture = acceptFutureWithExecutor(onWatermark(watermark, collector, coordinator),
        ignored -> watermarkStates.updateAggregateMetric(ssp, watermark));
    }

    return watermarkFuture;
  }
Aggregate the {@link WatermarkMessage} from each ssp into a watermark. Then call onWatermark() if
a new watermark exits.
@param watermarkMessage a {@link WatermarkMessage} object
@param ssp {@link SystemStreamPartition} that the message is coming from.
@param collector message collector
@param coordinator task coordinator
samza-core/src/main/java/org/apache/samza/operators/impl/OperatorImpl.java