源码执行流程图

类源码

@Public
public class BoundedOutOfOrdernessWatermarks<T> implements WatermarkGenerator<T> {

    /** 最大事件时间 */
    private long maxTimestamp;

    /** 最大延迟时间 */
    private final long outOfOrdernessMillis;

    /**
     * 
     *
     * @param maxOutOfOrderness 最大延迟时间
     */
    public BoundedOutOfOrdernessWatermarks(Duration maxOutOfOrderness) {
        checkNotNull(maxOutOfOrderness, "maxOutOfOrderness");
        checkArgument(!maxOutOfOrderness.isNegative(), "maxOutOfOrderness cannot be negative");

        this.outOfOrdernessMillis = maxOutOfOrderness.toMillis();

        // start so that our lowest watermark would be Long.MIN_VALUE.
        this.maxTimestamp = Long.MIN_VALUE + outOfOrdernessMillis + 1;
    }

    // ------------------------------------------------------------------------
    /*
    * 取出最大事件时间
    * 执行过程：比较 maxTimestamp 和 当前数据的事件时间，将最大时间赋给 maxTimestamp
    */
    @Override
    public void onEvent(T event, long eventTimestamp, WatermarkOutput output) {
        maxTimestamp = Math.max(maxTimestamp, eventTimestamp);
    }
    // 创建watermark，传入值：最大事件时间 - 最大延迟时间 -1
    @Override
    public void onPeriodicEmit(WatermarkOutput output) {

        output.emitWatermark(new Watermark(maxTimestamp - outOfOrdernessMillis - 1));
    }
}

原文出处：/show_1IX482qvtm3f.html