Flink源码阅读：Watermark机制

前面我们已经梳理了 Flink 状态和 Checkpoint 相关的源码。从本文开始，我们再来关注另外几个核心概念，即时间、Watermark 和窗口。
写在前面

在 Flink 中 Watermark 是用来解决数据乱序问题的，它也是窗口关闭的触发条件。对于 Watermark 的概念和用法还不熟悉的同学可以先阅读Flink学习笔记：时间与Watermark一文。下面我们进入正题，开始梳理 Watermark 相关的源码。
Watermark 定义

Watermark 的定义非常简单，它继承了 StreamElement 类，内部只有一个 timestamp 变量。

1. @PublicEvolving
2. public class Watermark extends StreamElement {
4.     /** The watermark that signifies end-of-event-time. */
5.     public static final Watermark MAX_WATERMARK = new Watermark(Long.MAX_VALUE);
7.     /** The watermark that signifies is used before any actual watermark has been generated. */
8.     public static final Watermark UNINITIALIZED = new Watermark(Long.MIN_VALUE);
10.     // ------------------------------------------------------------------------
12.     /** The timestamp of the watermark in milliseconds. */
13.     protected final long timestamp;
15.     /** Creates a new watermark with the given timestamp in milliseconds. */
16.     public Watermark(long timestamp) {
17.         this.timestamp = timestamp;
18.     }
20.     /** Returns the timestamp associated with this {@link Watermark} in milliseconds. */
21.     public long getTimestamp() {
22.         return timestamp;
23.     }
25.     // ------------------------------------------------------------------------
27.     @Override
28.     public boolean equals(Object o) {
29.         return this == o
30.                 || o != null
31.                         && o.getClass() == this.getClass()
32.                         && ((Watermark) o).timestamp == timestamp;
33.     }
35.     @Override
36.     public int hashCode() {
37.         return (int) (timestamp ^ (timestamp >>> 32));
38.     }
40.     @Override
41.     public String toString() {
42.         return "Watermark @ " + timestamp;
43.     }
44. }

复制代码

Watermark 处理过程

我们先来回顾一下 Watermark 的生成方法。

1. SingleOutputStreamOperator<Event> withTimestampsAndWatermarks = source
2.         .assignTimestampsAndWatermarks(
3.                 WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(20))
4.         );

复制代码

初始化

在定义 Watermark 的时候，我们调用 assignTimestampsAndWatermarks 方法。

1. public SingleOutputStreamOperator<T> assignTimestampsAndWatermarks(
2.         WatermarkStrategy<T> watermarkStrategy) {
3.     final WatermarkStrategy<T> cleanedStrategy = clean(watermarkStrategy);
4.     // match parallelism to input, to have a 1:1 source -> timestamps/watermarks relationship
5.     // and chain
6.     final int inputParallelism = getTransformation().getParallelism();
7.     final TimestampsAndWatermarksTransformation<T> transformation =
8.             new TimestampsAndWatermarksTransformation<>(
9.                     "Timestamps/Watermarks",
10.                     inputParallelism,
11.                     getTransformation(),
12.                     cleanedStrategy,
13.                     false);
14.     getExecutionEnvironment().addOperator(transformation);
15.     return new SingleOutputStreamOperator<>(getExecutionEnvironment(), transformation);
16. }

复制代码

这个方法接收了一个 WatermarkStrategy 参数，把它封装到 TimestampsAndWatermarksTransformation 中之后，就添加到 transformations 列表中了。在生成 StreamGraph 的过程中，会调用每个 transformation 的 transform 方法。

通过这个调用链路，创建出了 TimestampsAndWatermarksOperatorFactory，在初始化 StreamTask 时，会调用 TimestampsAndWatermarksOperatorFactory.createStreamOperator 方法来创建 TimestampsAndWatermarksOperator，并调用它的 open 方法。
在这个 open 方法中，主要是生成 timestampAssigner 和 watermarkGenerator。timestampAssigner 是用于提取时间戳，watermarkGenerator 是用于生成 Watermark。
生成完成之后注册了一个定时器，到指定时间后会调用 onProcessingTime 方法。

1. public void onProcessingTime(long timestamp) throws Exception {
2.     watermarkGenerator.onPeriodicEmit(wmOutput);
4.     final long now = getProcessingTimeService().getCurrentProcessingTime();
5.     getProcessingTimeService().registerTimer(now + watermarkInterval, this);
6. }

复制代码

这个方法的逻辑也很简单，先发送创建并发送 Watermark，然后再注册一个定时器。
发送 Watermark

我们以 BoundedOutOfOrdernessWatermarks 为例，它向下游发送了一个 Watermark，时间戳为 maxTimestamp - outOfOrdernessMillis - 1（maxTimestamp 是当前最大的事件时间戳，outOfOrdernessMillis 是我们定义的周期时间毫秒值）。随后在 WatermarkEmitter.emitWatermark 方法中，更新了当前 Watermark 的值。最后 RecordWriterOutput.emitWatermark 则是向下游广播当前的 Watermark。
下游处理

下游处理方法我们从 StreamOneInputProcessor.processInput 入手，先来看具体的调用链路。

在 inputWatermark 方法中，先是对 alignedSubpartitionStatuses 进行调整，alignedSubpartitionStatuses 这个变量主要是用来获取最小的 Watermark。最后调用了 findAndOutputNewMinWatermarkAcrossAlignedSubpartitions 方法。这个方法中，会获取到所有上游最小的 Watermark，如果它大于最近发送的一个 Watermark，就会向下游发送。

1. public void emitWatermark(Watermark watermark) throws Exception {
2.     watermarkGauge.setCurrentWatermark(watermark.getTimestamp());
3.     operator.processWatermark(watermark);
4. }

复制代码

这个发送方法中，调用了 operator.processWatermark，我们接着看这个处理方法。

在 tryAdvanceWatermark 方法中如果 Watermark 的时间大于 eventTimeTimersQueue 队列中头节点的时间，那么对 eventTimeTimersQueue 这个队列进行出队操作，这个操作意味着触发了窗口计算。
[code]public boolean tryAdvanceWatermark(        long time, InternalTimeServiceManager.ShouldStopAdvancingFn shouldStopAdvancingFn)        throws Exception {    currentWatermark = time;    InternalTimer timer;    boolean interrupted = false;    while ((timer = eventTimeTimersQueue.peek()) != null            && timer.getTimestamp()