private CompletionStage<Void> onWatermark(long watermark, MessageCollector collector, TaskCoordinator coordinator) {
    final long inputWatermarkMin;
    if (prevOperators.isEmpty()) {
      // for input operator, use the watermark time coming from the source input
      inputWatermarkMin = watermark;
    } else {
      // InputWatermark(op) = min { OutputWatermark(op') | op' is upstream of op}
      inputWatermarkMin = prevOperators.stream()
          .map(op -> op.getOutputWatermark())
          .min(Long::compare)
          .get();
    }

    CompletionStage<Void> watermarkFuture = CompletableFuture.completedFuture(null);
    if (currentWatermark < inputWatermarkMin) {
      // advance the watermark time of this operator
      currentWatermark = inputWatermarkMin;
      LOG.trace("Advance input watermark to {} in operator {}", currentWatermark, getOpImplId());

      final Long outputWm;
      final Collection<RM> output;
      final WatermarkFunction watermarkFn = getOperatorSpec().getWatermarkFn();
      if (watermarkFn != null) {
        // user-overrided watermark handling here
        output = (Collection<RM>) watermarkFn.processWatermark(currentWatermark);
        outputWm = watermarkFn.getOutputWatermark();
      } else {
        // use samza-provided watermark handling
        // default is to propagate the input watermark
        output = handleWatermark(currentWatermark, collector, coordinator);
        outputWm = currentWatermark;
      }

      if (!output.isEmpty()) {
        watermarkFuture = CompletableFuture.allOf(
            output.stream()
                .flatMap(rm -> this.registeredOperators.stream()
                    .map(op -> op.onMessageAsync(rm, collector, coordinator)))
                .toArray(CompletableFuture[]::new));
      }

      watermarkFuture =
          composeFutureWithExecutor(watermarkFuture, res -> propagateWatermark(outputWm, collector, coordinator));
    }

    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