Flink Watermark源码解析：WatermarkGenerator与窗口触发机制

做流处理这几年，被问得最多的一类问题，几乎都和 Flink 的 Watermark 有关。不是“Watermark 是什么”这种概念题，而是“我明明设置了 WatermarkGenerator，窗口为什么就是不触发”“水位线卡在某个值上一动不动”“并行度调大之后延迟突然涨了好几倍”。这些问题的共同点是：只看官方文档里的那几行示例代码，看不出答案，必须翻到 WatermarkGenerator 的实现里，看清楚它被谁调用、什么时候调用、调用之后那根水位线又经过了几层包装才流到下游。这篇内容就是把我自己读 Flink 源码的路径重新走一遍，从 WatermarkGenerator 这个只有两个方法的接口切入，一路跟到 TimestampsAndWatermarksOperator，再到多输入场景下的水位线裁决逻辑，中间穿插我自己写自定义生成器时踩过的坑和验证手段。适合已经能跑通 Flink 作业、但对事件时间和水位线心里没底的兄弟，也适合想动手改 WatermarkGenerator 但不确定副作用的人。

1. 先把问题摆到桌面上：为什么非要有这根水位线

1.1 从窗口触发的那个死结说起

假设你在做一个按分钟统计的实时看板，数据来自一堆客户端上报，每条记录自带一个发生时间戳。业务上你要的是“按照数据真正发生的时间，每 1 分钟聚一次”，也就是事件时间语义。现在问题来了：你要等到什么时候，才敢把 10:00 到 10:01 这一分钟的窗口算出来并输出结果？

这里有个绕不过去的死结。数据在网络里是会乱序的，10:00:59 的那条记录，可能比 10:00:03 的那条晚到十几秒。如果你一看到时间戳越过 10:01:00 就把窗口结算了，那后面姗姗来迟的记录就只能被丢掉或者重算。反过来，如果你一直等，等到“确定不会再有 10:01 之前的数据了”，那这个“确定”本身就需要一个外部信号——流处理系统自己没法凭空知道未来还有没有数据。

Watermark 就是这个外部信号的形式化表达。它是一条特殊的数据记录，只带一个时间戳，语义是：“事件时间小于等于这个时间戳的数据，我认为已经全部到齐了。”注意这里用的是“我认为”，不是“我保证”，这个措辞差别很关键，后面讲迟到数据的时候还会回到这一点。

所以整个机制的逻辑链条是：数据自带时间戳 → 用一个生成策略持续地抬升水位线 → 算子收到水位线后，把“已到期”的定时器（比如窗口结束时间）触发掉 → 结果输出。水位线的推进速度，直接决定了你的结果延迟和结果准确性之间的平衡点，而这个平衡点，就是 WatermarkGenerator 要负责的事情。

1.2 Watermark 在源码里到底长什么样

很多人第一次翻到org.apache.flink.streaming.api.watermark.Watermark这个类的时候都会有点意外，因为它实在是太简单了。它继承自StreamElement，和StreamRecord、WatermarkStatus是平级的，核心就只有一个long timestamp字段，外加一个常量MAX_WATERMARK = new Watermark(Long.MAX_VALUE)。

这个设计其实挺说明问题的。水位线在运行时就是一条普通元素，跟着数据记录一起在算子之间流动，走的是同一套网络传输和反压机制。它不是某种控制平面的元数据，也不是在某个中心节点维护的全局变量。这意味着两件事：第一，水位线会像数据一样被缓冲、被反压卡住，如果一个算子处理慢，它上游的水位线也会跟着堵；第二，水位线在跨算子传递时是有分区概念的，这一点在后面讲并行度的时候会展开。

另外一个容易被忽略的细节是Watermark类没有实现任何比较逻辑，也没有单调性校验。也就是说，框架层面并不会阻止你发出一个比上一条更小的水位线，虽然那样做在语义上是错误并且会导致下游定时器状态混乱。这个约束是靠WatermarkOutput那一层的包装来兜底的，而不是靠Watermark自己。理解这一点很重要：你写自定义生成器时，如果直接拿到WatermarkOutput就乱发，很可能不会立刻报错，而是过一段时间之后在某个窗口算子里看到莫名其妙的结果。

1.3 三条语义保证和它们各自的代价

把水位线的语义拆开，实际上有三条相互关联的保证在起作用。

第一条是单调性：同一个输出通道上的水位线只能增不能减。这条保证让下游可以安全地用一个long变量记住“当前水位线”，不用做任何比较。代价是如果一个分区长时间没有数据，它的水位线就不动，会把整个下游卡住——这直接催生了空闲检测（idleness）机制。

第二条是全序到偏序的转换：单个分区内的水位线是单调的，但多个上游分区之间没有顺序关系。下游算子必须把它们合并成一个，合并规则是取最小值。取最小值是为了保证“所有上游都说这个时间点之前的数据到齐了”才成立，任何一路没到齐，就不能推进。代价是只要有一路慢，全局都慢，这是水位线最容易出问题的地方。

第三条是不保证不丢数据：水位线只表达“我认为到齐了”，不表达“我保证到齐了”。越过水位线之后到达的数据就是迟到数据，默认会被丢弃。代价是你要么加大乱序容忍度换取准确性、牺牲延迟，要么开启allowedLateness让窗口多留一会儿，要么用侧输出把迟到数据接出来单独处理。

这三条放在一起看，Watermark 的整个设计取向就清楚了：它不追求绝对正确，它追求的是在有限延迟内给出一个可以量化的近似结果，把“要多准”这个决策权交给使用者。而这个决策权的具体承载物，就是接下来要拆的WatermarkGenerator。

2. WatermarkGenerator 接口：两个方法背后的调用时序

2.1 onEvent 与 onPeriodicEmit 分别被谁调用

打开org.apache.flink.api.common.eventtime.WatermarkGenerator，你会看到一个特别克制的接口定义，只有两个方法：onEvent(T event, long eventTimestamp, WatermarkOutput output)和onPeriodicEmit(WatermarkOutput output)。没有 init，没有 close，没有获取当前水位线的 getter。这种极简设计是有意为之的，接口越窄，框架对实现的假设就越少。

onEvent的调用时机很明确：每处理一条数据就调一次，在TimestampsAndWatermarksOperator.processElement里，紧跟在时间戳提取之后。注意传入的eventTimestamp是已经由TimestampAssigner提取好的值，不是你原始数据里的字段，所以如果你在onEvent里再去解析一遍字段，纯属浪费。

onPeriodicEmit的调用时机就有讲究了。它是由算子内部基于处理时间注册的一个定时器周期性触发的，间隔由pipeline.auto-watermark-interval控制，默认 200 毫秒。这里的关键词是“处理时间”——也就是说，即使你的数据流完全静止、一条数据都不来，只要这个定时器还在跑，onPeriodicEmit就会被调用。这解释了一个常见困惑：“我的流里没数据了，为什么下游还在收到水位线？”答案是周期型生成器在空转。

还有一个隐蔽的细节：当auto-watermark-interval被设置为 0 的时候，周期定时器根本不会被注册，onPeriodicEmit永远不会被调用。这就是为什么纯打点型（punctuated）生成器必须把这个间隔设为 0，否则就是白白浪费一次回调。

2.2 周期型和打点型，本质是同一接口的两种用法

官方文档把水位线生成分成“周期型”和“打点型”两类，但从源码角度看，这两者并没有对应的两个接口，它们只是同一个WatermarkGenerator接口的两种填充方式。

周期型的做法是：onEvent里只更新内部状态（比如记录见过的最大时间戳），真正的emitWatermark放在onPeriodicEmit里。好处是发水位线的频率和数据量解耦，吞吐高的时候不会因为每条数据都发一次水位线而把网络打满。坏处是水位线的推进有最多一个 interval 的延迟，精细控制做不到。

打点型的做法反过来：onEvent里判断某个条件（比如遇到一个特殊标记事件、或者跨过了某个整数边界），条件满足就立刻output.emitWatermark(...)，onPeriodicEmit留空。好处是响应及时，语义上可以和业务事件绑死。坏处是如果条件设计得不好，可能出现长时间不发水位线导致下游窗口不触发，或者发得太频繁造成额外开销。

实际项目里我基本都用周期型，因为绝大多数数据源没有可靠的“分片结束”标记，硬凑一个打点条件反而容易出错。只有在处理那种带明确批次结束标记的数据（比如文件导入、MQ 里的批次完成消息）时，打点型才有明显优势。

2.3 WatermarkOutput：被包装了一层的那道闸门

WatermarkOutput这个接口名字看着像输出端，实际用起来更像一道闸门。它只有一个emitWatermark(Watermark watermark)方法和一个markIdle()（新版是emitWatermarkStatus(WatermarkStatus)）。

为什么需要这层抽象？因为生成器发出去的水位线，不一定会原封不动地流到网络上去。在算子内部，这个output可能是一个WatermarkOutputMultiplexer.Output，也可能是一个带着空闲状态跟踪的包装。最典型的是WatermarkOutputMultiplexer的场景：当一个算子的输出被链式（chained）连接到多个下游输入时，生成器只发一次水位线，多路复用器要把它记录下来，然后对每一路下游分别计算“应该看到的水位线”，再分别转发。

这里有个很容易忽略的点：WatermarkOutputMultiplexer在合并多路输出时取的也是最小值。也就是说，如果你在一个ProcessFunction里做了分流，两路下游的处理速度不一样，慢的那一路会拖住快的那一路看到的水位线吗？答案是会，因为多路复用器是取所有 active 输出的最小值来算出“组合水位线”的。这是从正确性出发的保守设计，实际表现就是分流之后整体水位线推进变慢。我在一个双流 join 的项目里被这个坑过一次，排查了半天才意识到是其中一路下游反压导致水位线整体停滞。

3. 沿源码往下走：从一个接口到真正的水位线

3.1 BoundedOutOfOrdernessWatermarks 到底做了什么

几乎所有人第一个接触的实现都是BoundedOutOfOrdernessWatermarks，也就是WatermarkStrategy.forBoundedOutOfOrderness(Duration)背后那个类。它的代码短得可怜，但每一行都有讲究。

public class BoundedOutOfOrdernessWatermarks<T> implements WatermarkGenerator<T> { /** The maximum timestamp encountered so far in the stream. */ private long maxTimestamp; /** The maximum out-of-orderness that this watermark generator assumes. */ private final long outOfOrdernessMillis; 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; } @Override public void onEvent(T event, long eventTimestamp, WatermarkOutput output) { maxTimestamp = Math.max(maxTimestamp, eventTimestamp); } @Override public void onPeriodicEmit(WatermarkOutput output) { output.emitWatermark(new Watermark(maxTimestamp - outOfOrdernessMillis - 1)); } }

先看构造函数里那个初始值Long.MIN_VALUE + outOfOrdernessMillis + 1。第一眼看上去很奇怪，为什么要加个outOfOrdernessMillis + 1？原因是这一行同时防了两件事。一是防止下溢：如果maxTimestamp从Long.MIN_VALUE开始，onPeriodicEmit里算Long.MIN_VALUE - outOfOrdernessMillis - 1就会直接溢出成一个巨大的正数，发出去一条Long.MAX_VALUE级别的水位线，下游所有窗口瞬间全部触发。二是保证初始水位线恰好落在Long.MIN_VALUE上，这样“还没收到任何数据”的状态不会误触发任何定时器。

再看onPeriodicEmit里那个- 1。这个才是最容易看漏的地方。假设乱序容忍度是 0（AscendingTimestampsWatermarks的情况），那么发出的水位线就是maxTimestamp - 1，而不是maxTimestamp。为什么要减这个 1？因为 Watermark 的语义是“小于等于该时间戳的数据已经到齐”，而窗口的触发条件是watermark >= window.maxTimestamp()，其中window.maxTimestamp()对于时间窗口来说是endTime - 1。如果不减 1，那么当水位线正好等于maxTimestamp时，maxTimestamp本身那条数据对应的窗口边界可能还没到，就会出现“水位线到了但窗口没触发”或者“窗口提前触发”的偏差。这个 1 是边界对齐的修正项，不是拍脑袋。

3.2 时间戳提取器和生成器的组装过程

单独一个WatermarkGenerator是不能工作的，它必须和一个TimestampAssigner一起打包成WatermarkStrategy。而这个打包动作发生在作业图构建阶段，不是在运行时。

WatermarkStrategy的实现类WatermarkStrategy.WithTimestampsWatermarks里有个createWatermarkGenerator方法，它会调用WatermarkGeneratorSupplier.createWatermarkGenerator(context)。这个context是WatermarkGeneratorSupplier.Context，里面能拿到MetricGroup、StreamConfig、ProcessingTimeService之类的运行时资源。这就是为什么你可以在createWatermarkGenerator里拿到 metric 分组、注册自定义指标——很多生产项目就是靠这个把水位线延迟做成了监控面板。

WatermarkStrategy<MyEvent> strategy = WatermarkStrategy .<MyEvent>forBoundedOutOfOrderness(Duration.ofSeconds(5)) .withTimestampAssigner((event, recordTimestamp) -> event.getEventTime()) .withIdleness(Duration.ofSeconds(30));

这段代码里的链式调用顺序其实有讲究。withTimestampAssigner返回的是一个新对象，withIdleness也是。每个方法都在原来的策略上加一层包装，最后交给TimestampsAndWatermarksOperator的是一个组合好的策略。如果你的TimestampAssigner内部有状态（比如需要查维表），务必保证它是可序列化的，因为整个策略对象会被序列化进作业图，在 TaskManager 端反序列化执行。我见过有人在匿名内部类里捕获了一个数据库连接池，作业提交直接失败。

3.3 TimestampsAndWatermarksOperator 的 processElement 走读

真正干活的是org.apache.flink.streaming.runtime.operators.TimestampsAndWatermarksOperator。它的processElement是一段特别紧凑的代码，但每一步都不能少。

@Override public void processElement(StreamRecord<T> element) throws Exception { final long newTimestamp = userFunction.extractTimestamp( element.getValue(), element.hasTimestamp() ? element.getTimestamp() : Long.MIN_VALUE); output.collect(element.replace(element.getValue(), newTimestamp)); userFunction.onEvent(element.getValue(), newTimestamp, output); }

注意执行顺序：先提取时间戳，然后立刻把记录（带新时间戳）往下游发出去，最后才调用onEvent。这个顺序非常关键。它意味着当前这条数据本身是在水位线更新之前就发出去的，不会出现“数据还没发，水位线先跑了，导致这条数据被判为迟到”的荒谬情况。

还有一个点是element.hasTimestamp() ? element.getTimestamp() : Long.MIN_VALUE。如果上游没有给记录设置时间戳，这里会传Long.MIN_VALUE给提取器。你的TimestampAssigner第二个参数拿到Long.MIN_VALUE的时候不要惊讶，这只是一个“没有上游时间戳”的标记。如果你在提取器里对第二个参数做算术运算，会直接溢出，这是个很典型的低级错误。

processWatermark就简单得多，它把上游来的水位线直接output.emitWatermark(mark)转发下去。但是注意，默认情况下TimestampsAndWatermarksOperator会优先使用自己生成的水位线，而不是上游的。这就有个容易搞混的场景：如果你在一个已经带事件时间的流上再套一层assignTimestampsAndWatermarks，那么下游看到的水位线是由新生成的策略决定的，上游那条水位线在这里被“截断”了。我见过有人重复分配时间戳，导致水位线推进速度完全不符合预期，最后发现问题出在中间多做了一次 assign。

3.4 StatusWatermarkValve：多输入场景下的合成逻辑

单输入的水位线就是一条直线往下传，没什么好说的。真正复杂的是多输入，比如connect、union、多路 join。这时候框架要解决一个问题：两个输入各自有各自的水位线，下游算子只有一个“当前水位线”的概念，怎么合并？

答案是org.apache.flink.streaming.runtime.io.StatusWatermarkValve。它的核心逻辑是维护一个StreamStatus数组，每个输入通道一个槽位，记录该通道的最新水位线和是否处于活跃状态。每次某个通道的水位线更新，它就重新计算一次：取所有活跃通道的水位线最小值，如果这个最小值比当前已发出的水位线大，就往下游发一条。

这里的关键词是“活跃”。如果一个通道长时间没有数据，它的水位线会停在原地，把整个最小值卡死。StatusWatermarkValve提供的解法是通道状态标记：把没数据的通道标记为 IDLE，计算最小值的时候跳过它。这就是withIdleness在底层真正的落点。

4. 自己写一个 WatermarkGenerator：从需求到验证

4.1 为什么内置实现有时不够用

BoundedOutOfOrdernessWatermarks的问题在于它的乱序容忍度是一个静态常量。你设 5 秒，那它就永远是 5 秒。但真实数据流的乱序程度是会变的：白天高峰期网络拥塞、乱序严重，凌晨低谷时几乎有序。固定值要么在高峰期不够用（丢数据），要么在低谷期太保守（白白增加延迟）。

一个更聪明的做法是让生成器自己观察并调整。思路不复杂：在onEvent里，如果新来的数据时间戳小于当前maxTimestamp，说明这条数据迟到了，迟到幅度就是maxTimestamp - eventTimestamp。把这个幅度记录下来，用某种方式聚合成一个“近期观察到的最坏乱序量”，然后在onPeriodicEmit里用这个动态值代替静态值。

聚合方式我试过两种。一种是简单的滑动窗口最大值，维护一个固定长度的时间桶队列，取最近 N 个桶里的最大值。另一种是指数衰减，每来一条数据就把当前估计值乘一个衰减系数再和新观测值取大。前者的行为更可预测、更容易解释给业务方，后者实现更简单、内存占用恒定。生产上我倾向第一种，因为“最近一分钟内最坏乱序是 8 秒”这句话能直接跟人解释，而衰减系数不太好沟通。

4.2 完整实现代码

下面这个版本用的是固定时间桶的滑动最大值，桶宽和桶数都可以通过构造函数指定，实际相当于“最近bucketSize * numBuckets时间内的最大乱序观测值”。

import org.apache.flink.api.common.eventtime.Watermark; import org.apache.flink.api.common.eventtime.WatermarkGenerator; import org.apache.flink.api.common.eventtime.WatermarkOutput; public class AdaptiveOutOfOrdernessWatermarks<T> implements WatermarkGenerator<T> { private final long bucketWidthMs; private final long[] buckets; private int currentBucketIndex; private long currentBucketStart; private long maxTimestamp = Long.MIN_VALUE; private boolean initialized = false; private final long floorLatenessMs; private final long ceilingLatenessMs; public AdaptiveOutOfOrdernessWatermarks(long bucketWidthMs, int numBuckets, long floorLatenessMs, long ceilingLatenessMs) { this.bucketWidthMs = bucketWidthMs; this.buckets = new long[numBuckets]; this.floorLatenessMs = floorLatenessMs; this.ceilingLatenessMs = ceilingLatenessMs; } @Override public void onEvent(T event, long eventTimestamp, WatermarkOutput output) { if (!initialized) { initialized = true; maxTimestamp = eventTimestamp; currentBucketStart = eventTimestamp - (eventTimestamp % bucketWidthMs); return; } if (eventTimestamp > maxTimestamp) { maxTimestamp = eventTimestamp; } else { long lateness = maxTimestamp - eventTimestamp; rotateBucketsIfNeeded(eventTimestamp); int idx = currentBucketIndex; buckets[idx] = Math.max(buckets[idx], lateness); } } private void rotateBucketsIfNeeded(long eventTimestamp) { long bucketStart = eventTimestamp - (eventTimestamp % bucketWidthMs); long delta = (bucketStart - currentBucketStart) / bucketWidthMs; if (delta <= 0) { return; } int steps = (int) Math.min(delta, buckets.length); for (int i = 0; i < steps; i++) { currentBucketIndex = (currentBucketIndex + 1) % buckets.length; buckets[currentBucketIndex] = 0L; } currentBucketStart = bucketStart; } private long currentLatenessBudget() { long max = 0L; for (long b : buckets) { if (b > max) { max = b; } } long budget = Math.max(max * 2, floorLatenessMs); return Math.min(budget, ceilingLatenessMs); } @Override public void onPeriodicEmit(WatermarkOutput output) { if (!initialized) { return; } output.emitWatermark(new Watermark(maxTimestamp - currentLatenessBudget() - 1)); } }

几点说明。max * 2这个系数是留的余量，因为观测到的乱序只是“已经发生的”，未来可能更糟，翻倍是个粗糙但有效的安全边际。floorLatenessMs防止在数据完全有序时预算降到 0 导致水位线过于激进，ceilingLatenessMs防止某次异常乱序把水位线永久拖住——注意由于桶会滚动，异常值最终会被挤出窗口，但设个上限更保险。

用的时候这样接：

DataStream<MyEvent> stream = env.addSource(source) .assignTimestampsAndWatermarks( WatermarkStrategy .<MyEvent>forGenerator(ctx -> new AdaptiveOutOfOrdernessWatermarks<>(10_000L, 6, 1_000L, 60_000L)) .withTimestampAssigner((e, ts) -> e.getEventTime()) );

forGenerator接收的是一个WatermarkGeneratorSupplier，注意这里传的是 lambda，每次算子实例化时会调用一次，所以每个 subtask 都有自己独立的生成器实例，状态互不共享。这一点很重要：生成器的状态是 per-subtask 的，不是全局的。如果你的数据分布不均匀，不同 subtask 观察到的乱序程度会不一样，发出的水位线也会不一样，最后在下游合并时取最小值，整体效果由最慢的那个决定。

4.3 参数怎么定：从业务 SLA 反推

定参数这件事，不要从“我觉得乱序大概几秒”开始，而要从业务能接受的延迟倒推。假设业务要求数据在事件发生 30 秒内出现在结果里，那么总延迟预算 30 秒里要扣掉：网络传输（通常 1 到 3 秒）、算子处理（视逻辑复杂度，留 2 到 5 秒）、checkpoint 造成的短暂阻塞（如果开了 exactly-once，留 1 到 2 秒）、周期发射间隔（最多 200 毫秒，可忽略）。剩下的才是留给乱序容忍度的空间，大概 20 秒左右。

然后拿这个预算去比对实际观测到的乱序分布。方法很粗暴但有效：先设一个很大的容忍度（比如 60 秒），跑一天，同时把每条数据的迟到幅度打到日志或指标里，最后看 P99 是多少。如果 P99 是 6 秒，那设 8 到 10 秒就够了，不用把 20 秒全用掉。剩下的预算可以作为 buffer 应对突发。

提示：不要用平均值来定这个参数。乱序分布是典型的长尾分布，平均值毫无意义，一定要看高分位数。我见过用平均值设成 2 秒结果丢掉 8% 数据的案例。

4.4 怎么确认水位线真的按预期在推进

代码写完只是第一步，验证才是重点。最直接的手段是 Web UI 上算子详情页里的currentInputWatermark和currentOutputWatermark两个指标，它们会显示当前 subtask 看到和发出的水位线数值。但这两个数只有瞬时值，看不出趋势。

更好的做法是自己注册一个 gauge。在自定义生成器的createWatermarkGenerator里能拿到MetricGroup，可以这样挂：

WatermarkStrategy.<MyEvent>forGenerator(ctx -> { MetricGroup metrics = ctx.getMetricGroup(); AdaptiveOutOfOrdernessWatermarks<MyEvent> gen = new AdaptiveOutOfOrdernessWatermarks<>(10_000L, 6, 1_000L, 60_000L); metrics.gauge("watermarkLagMs", () -> System.currentTimeMillis() - gen.currentWatermark()); return gen; })

注意currentWatermark()这个方法在上面那份代码里我没写，实际用的时候需要补一个返回maxTimestamp - currentLatenessBudget() - 1的方法出来。有了这个指标，你就能在监控面板上直接看到“水位线落后当前系统时间多少毫秒”随时间的变化曲线，调参的时候盯着这条曲线看，比猜靠谱得多。

5. 踩坑实录：那些文档里不会写的问题

5.1 水位线卡住不动的四类原因

这是最高频的问题，我按遇到频率排了个序。

第一类是某个并行度没数据。假设 source 并行度是 8，其中 3 个 subtask 分到的分区暂时没有数据写入，它们的生成器就永远停在初始状态，发出的水位线是Long.MIN_VALUE。下游算子合并时取最小值，结果就是Long.MIN_VALUE，所有窗口都不触发。解法是withIdleness，把空闲通道标记出来跳过。

第二类是反压传导。水位线和数据走同一条通道，如果某个下游算子的输入缓冲区满了，水位线也发不出去。这种情况在 Web UI 上会看到明显的反压标记，但如果你只盯着水位线指标看，会觉得莫名其妙地停了。这时候要去查下游算子的处理耗时，而不是怀疑生成器。

第三类是时间戳单位搞错。TimestampAssigner返回的必须是毫秒。如果你的数据源给的是秒级时间戳（10 位数字），直接返回就会得到一个 1970 年附近的时间，水位线永远在很低的数值徘徊，窗口永远不触发。这个问题在测试环境特别隐蔽，因为数据量小的时候你可能根本没注意到窗口没出结果。

第四类是生成器里的状态没被正确更新。比如你在onEvent里忘了更新maxTimestamp，只在某些分支更新，那水位线就会一直停在那里。自定义生成器一定要保证onEvent是无条件更新状态的。

5.2 空闲检测到底该怎么配

withIdleness(Duration)的参数是“多久没数据就标记为空闲”。这个值不能设得太小，否则正常的处理抖动会被误判为空闲，导致水位线在不该推进的时候推进，丢掉本该等待的数据。也不能设得太大，否则空闲分区的拖累时间会很长。

我的经验值是：这个值应该大于数据到达的典型间隔的 3 到 5 倍。比如你的分片数据平均每 2 秒来一条，那设 10 秒比较合适。如果数据到达本身就很稀疏（比如每 5 分钟一条），那这个机制其实不太适用，你可能需要考虑换一种分片策略，或者干脆不用事件时间窗口。

还有一个坑：withIdleness只对输入做空闲标记，它标记的是“这个输入通道空闲”。但如果你有多个 source，其中一个 source 整体停了，它的空闲标记要能传播到下游才行。这条路径在不同版本里行为有过调整，建议实际测试一下你的版本在“一路 source 完全停止”时的表现，别假设它一定能正常工作。

5.3 迟到数据的三种处置方式对比

水位线越过之后到达的数据，默认是被丢弃的。这在很多场景下不可接受，Flink 提供了三条路。

第一种是加大乱序容忍度。这是最省事的，代价是整体延迟变大。适合延迟不敏感、准确性要求高的场景。

第二种是开启允许迟到。在窗口上调用allowedLateness(Duration)，窗口触发之后不会立刻销毁，而是保留一段时间，期间每来一条迟到数据就重新触发一次计算。代价是每个窗口要额外维护一份状态，窗口越多内存压力越大。要注意这会带来“同一个窗口多次输出”的行为，下游必须能处理重复结果。

第三种是侧输出迟到数据。用sideOutputLateData(OutputTag)把迟到数据分流出去，单独做补偿处理。这是最灵活的方式，适合需要精确对账的场景。代价是你得为这条侧流写额外的处理逻辑。

实际项目里我经常混用：窗口上开一个较小的allowedLateness（比如 30 秒）覆盖绝大多数情况，同时挂一个侧输出兜住极端迟到数据，侧流写入一个单独的表，离线任务再和主结果做一次对账合并。这样既不牺牲太多实时性，也不会真的丢数据。

5.4 自定义生成器的两个隐蔽陷阱

第一个陷阱是每 subtask 独立状态带来的不一致。前面提过，生成器实例是 per-subtask 的。如果你的 source 数据分布有倾斜，某个 subtask 拿到的时间戳范围比另一个宽，它们发出的水位线差异可能很大。下游合并取最小值，整体进度由最慢的那个决定。这在调参时会造成困扰：你在某个 subtask 上看到水位线推进很快，但整体没动，会误以为是别的问题。排查时一定要同时看所有 subtask 的水位线，而不是只看一个。

第二个陷阱是在 onEvent 里直接 emit 水位线。技术上完全可行，但会引入一个微妙的问题：这条水位线是在处理当前数据的中间发出的，而当前数据的处理还没结束（虽然数据本身已经 collect 出去了）。如果下游有多个输入，这条早发的水位线可能先于其他输入的数据到达，造成短暂的“水位线跳变”。打点型生成器必须这么做，但如果做周期型，一定要忍住别在onEvent里 emit。

6. 几个容易被忽略但影响很大的工程细节

6.1 时间戳精度与边界的那点事

前面讲- 1的时候提了一句，这里展开说。窗口的触发条件在源码里是timerService.registerEventTimeTimer(window.maxTimestamp())，而Window.maxTimestamp()的默认实现是endTime - 1。对于[10:00:00, 10:01:00)这个窗口，maxTimestamp()是10:00:59.999，也就是10:00:59加上 999 毫秒——在毫秒精度的long表示里就是endTime - 1。

所以要让这个窗口触发，水位线必须>= 10:00:59.999。而BoundedOutOfOrdernessWatermarks发出的水位线是maxTimestamp - outOfOrderness - 1。假设乱序容忍度 0，一条时间戳正好是10:00:59.999的数据到达，发出的水位线是10:00:59.998，窗口不触发。要等下一条时间戳>= 10:01:00.000的数据到达，水位线变成10:00:59.999，窗口才触发。这就是为什么“以最大时间戳结尾的那批数据所在的窗口”需要额外一条后续数据才能关闭，这个行为经常让人困惑。

如果你的时间戳精度是秒（比如long存的是秒），那么所有时间戳都是整秒，maxTimestamp()是endTime - 1，你会发现窗口永远差一点。这种情况下必须把时间戳乘以 1000 转成毫秒，别无他法。

6.2 并行度、分区与水位线传播的关系

水位线在算子之间是广播的，也就是说上游每个 subtask 发出的水位线，会被发送到下游所有 subtask。下游每个 subtask 收到 N 条水位线（N 是上游并行度），合并取最小值。

这个机制带来两个后果。一是水位线的传输量随并行度线性增长，虽然每条只有几十字节，但高频发射还是有开销，这也是为什么周期型的默认间隔是 200 毫秒而不是 10 毫秒。二是木桶效应特别明显：上游并行度越大，越有可能出现某一两个慢的分区，把整体水位线拖住。

所以调大并行度的时候要留意水位线的表现。我在一个项目里把 source 并行度从 4 调到 16，吞吐上去了，但端到端延迟涨了将近一倍，原因就是 16 个分区里总有那么一两个分区数据稀疏，withIdleness配得又不够激进，水位线整体推进被拖慢。后来把 idleness 从 30 秒调到 8 秒才恢复正常。

6.3 checkpoint 恢复之后水位线在哪里

这是个容易让人心里没底的问题。WatermarkGenerator的内部状态（比如maxTimestamp）默认是不参与 checkpoint 的。因为它不是CheckpointedFunction，也没有实现任何快照接口。这意味着作业从 checkpoint 恢复之后，生成器的状态是全新的，maxTimestamp会被重置到初始值。

那会发生什么？恢复后的第一批数据会把maxTimestamp重新推上去，然后onPeriodicEmit发出一批水位线。由于恢复后的水位线从很小的值开始爬升，下游的窗口定时器（这些定时器是注册在 keyed state 里的，会随 checkpoint 恢复）在恢复后要等到水位线重新爬过它们触发点才会执行。

如果你的 checkpoint 间隔比较长，或者恢复后数据有断档，可能出现“恢复后一段时间内所有窗口都不触发，然后突然集中触发一大批”的现象。这个行为本身是语义正确的（at-least-once 或者 exactly-once 都允许重算），但如果你的下游对突发流量敏感，需要提前做容量准备。

注意：如果你确实需要生成器状态参与恢复（比如那个自适应乱序估计器，恢复后重新学习会有个冷启动期），可以自己实现CheckpointedFunction，在initializeState里恢复maxTimestamp和桶数组。不过要小心，恢复的maxTimestamp如果比当前实际数据的时间戳大很多，会导致恢复后立刻把水位线推得很高，把新数据全判成迟到。这个权衡得自己拿捏。

6.4 关于 final watermark 的一点观察

算子关闭的时候，默认会补发一条Watermark.MAX_WATERMARK，目的是让下游所有还挂着的窗口立刻结算并输出，避免作业停止时丢掉最后一批结果。这个行为在不同版本之间有过调整，有些版本可以通过参数关掉。

我的建议是自己动手验证一下你用的版本到底是什么行为，方法是写一个最简单的本地作业：一个 source 发几条带时间戳的数据，一个滑动窗口做聚合。跑完之后看输出，如果最后那批本该被窗口持有的数据被输出出来了，说明 final watermark 起了作用。这个验证十分钟就能做完，但能帮你避免很多“作业停了结果少了一截”的困惑。

另外提醒一句，如果你在管道里串了多个assignTimestampsAndWatermarks（虽然不推荐，但有时为了在不同阶段用不同策略会这么干），最终水位线是被最后一个生成器决定的，中间那些策略的 final watermark 行为可能互相影响，测试的时候要把整条链都跑通。

我自己在这个方向上折腾了挺久，最大的体会是：Watermark 的难点从来不在概念，概念三句话就能讲完；难点在于它是一条贯穿数据源、算子链、网络传输、状态恢复的完整链路，任何一个环节出问题，表现都是“窗口不触发”这一个症状。所以排查的时候不要只盯着生成器看，要从数据源的时间戳一路往下捋，看每个 subtask 的输入水位线、输出水位线、下游的合并结果，一级一级对，通常十分钟内就能定位。至于参数怎么调，别信任何“最佳实践”里给的固定数值，那些数字都是在特定数据分布下测出来的，你的数据分布不一样，答案就不一样，老老实实打点、观察、调、再观察，这才是唯一靠谱的路子。