private CompletionStage<Void> propagateWatermark(Long outputWm, MessageCollector collector, TaskCoordinator coordinator) {
    CompletionStage<Void> watermarkFuture = CompletableFuture.completedFuture(null);

    if (outputWm != null) {
      if (outputWatermark < outputWm) {
        // advance the watermark
        outputWatermark = outputWm;
        LOG.debug("Advance output watermark to {} in operator {}", outputWatermark, getOpImplId());
        watermarkFuture = CompletableFuture.allOf(
            this.registeredOperators
                .stream()
                .map(op -> op.onWatermark(outputWatermark, collector, coordinator))
                .toArray(CompletableFuture[]::new));
      } else if (outputWatermark > outputWm) {
        LOG.warn("Ignore watermark {} that is smaller than the previous watermark {}.", outputWm, outputWatermark);
      }
    }

    return watermarkFuture;
  }
A watermark comes from an upstream operator. This function decides whether we should update the
input watermark based on the watermark time of all the previous operators, and then call handleWatermark()
to let the inherited operator to act on it.
@param watermark incoming watermark from an upstream operator
@param collector message collector
@param coordinator task coordinator
samza-core/src/main/java/org/apache/samza/operators/impl/OperatorImpl.java