Flink Watermark空闲检测：withIdleness原理、版本差异与避坑

上个月帮一个团队看一个双流合并的任务，现象特别典型：两个流都在正常消费，checkpoint 一次不落，日志干干净净，但下游窗口就是不吐结果。打开 Web UI 的 Watermarks 面板一看，一个 task 的水位在一路往上走，另一个卡在初始值一动不动。最后定位到的原因也很朴素——那条卡住的通道已经很久没有新数据进来了，而 Flink 在多通道汇总水位时，仍然把这条“安静的”通道算进最小值里，于是整个算子的水位被一条不下蛋的腿死死拽住。

这就是WatermarkStrategy.withIdleness要解决的事。它做的事情听起来很小：给数据源加一个“空闲检测”，超过阈值没数据就把它标记成 idle，让它暂时退出最小水位的比较。但要真正用对、用稳，光会写一行.withIdleness(Duration.ofMinutes(1))远远不够——你得知道这个标记是谁发的、发到哪一层、谁来消费、恢复的时候会发生什么，以及它在不同版本里的实现位置其实变过一次。

这篇内容面向的是已经在用事件时间、被水位卡住过的同学，也面向想顺着源码把 watermark 传播链路捋一遍的同学。我会从最小水位语义讲起，把WatermarkStrategy的包装逻辑、IdlenessTimer的判定细节、markIdle之后的水位合并链路一条条拆开，再给一个能肉眼看到 idle 生效的完整 Demo，最后把几个真踩过的坑摆出来。源码部分我会保留主干、删掉日志和断言，读起来不至于被噪音淹没。

1. 一条不推进水位的通道，能把整个窗口拖死

先把这个问题的物理本质说清楚，不然后面看源码容易飘。事件时间窗口的触发条件只有一句话：当前算子的水位越过了窗口的结束时间。而水位不是某一个数据源说了算，它是多个输入通道协商出来的结果。

1.1 最小水位语义决定了“最慢的那条腿”

在 Flink 里，一个算子如果有多个输入通道（两个上游算子连过来、或者一个上游的多个并行子任务连过来），它维护的水位取所有通道水位的最小值。这个设计没有商量余地：因为水位语义是“时间戳小于等于 W 的数据不会再来”，只要有一条通道还没走到 W，你就不能断言全局走到了 W。

打个生活化的比方：一个班级集合出发，老师必须等最慢的那个学生到齐才能发车。这个规则本身没问题，问题出在“某个学生请假了，但他没告诉老师”。老师就一直等，队伍就一直不发车。withIdleness干的事，就是让请假的同学举一下手。

对应到代码里，多个通道的合并发生在StatusWatermarkValve这个组件中。它内部维护一个通道状态数组，每个元素记录两件事：这条通道当前的 watermark 值，以及这条通道当前是否处于对齐（active）状态。每次收到新的 watermark，它会更新对应通道的值，然后重新在所有处于对齐状态的通道里取最小值往下游发。

这里的关键词是“处于对齐状态”。一条通道被标记为 idle 之后，它的 watermark 值就不再参与最小值计算了——注意，是退出比较，不是把它的 watermark 设成Long.MAX_VALUE。这两种做法的区别很大：设成MAX_VALUE意味着“这条通道已经走完了所有时间”，一旦它后面又来数据，水位就得往回退，而水位是严格单调的，退不回去，只能被丢弃，这就是数据丢失的经典成因。退出比较则是安全的，通道恢复时把它的真实水位重新纳入比较即可。

1.2 一个只有两个并行度的复现场景

我见过的最容易复现的场景是这样的：数据源并行度设为 2，但上游的数据分布严重倾斜，或者某个分区被消费完了、某个分区暂时没有新消息。子任务 0 每秒发一条数据，子任务 1 一条都不发。

这时候你去看下游窗口算子的输入，它有两条 channel。channel 0 的水位在涨，channel 1 的水位停在初始值（Long.MIN_VALUE + 1）。取最小值，结果就是初始值。窗口的结束时间无论设多小，都不可能被“越过”，于是：

• 窗口永远不触发，哪怕窗口里已经攒了几万条数据；
• 注册的 event-time 定时器永远不响，如果你用KeyedProcessFunction做过期清理，清理逻辑等于失效；
• 窗口状态只增不减，checkpoint 体积一路膨胀，跑上几天就开始出现 checkpoint 超时；
• 最迷惑人的是，所有监控指标看起来都正常：吞吐在涨，CPU 不高，checkpoint 成功。

这个组合——吞吐正常、checkpoint 正常、业务静默——基本就是水位问题的指纹。下次遇到，先去 Watermarks 面板看一眼，比翻日志快得多。

1.3 卡住的表象：checkpoint 正常，业务静默

有人会问，为什么 checkpoint 正常反而成了干扰项？因为 checkpoint 的触发靠的是 checkpoint coordinator 的定时器，跟水位完全无关；而窗口状态的快照只要算子还在跑就能做。所以水位卡死不会影响 checkpoint 的成功率，只会让 checkpoint 越来越大。等哪天状态大到超过阈值，才会以“checkpoint 超时”的形式暴露出来，那时候离故障发生已经过去很久了。

我在排查时会顺手看一眼 checkpoint size 的历史曲线，如果它是单调爬升、锯齿不明显（正常应该有“涨-落”的锯齿，因为窗口触发后会清状态），那就基本能确认窗口没触发。这个信号比看水位面板更早出现，适合做预警。

2. 拆开 withIdleness：WatermarkStrategy 到底包了什么

WatermarkStrategy是一个接口，同时继承了TimestampAssignerSupplier和WatermarkGeneratorSupplier。它本身是个“工厂的工厂”：你给我上下文，我给你造一个时间戳分配器和一个水位生成器。withIdleness没有改变这个契约，它只是套了一层装饰器。

2.1 装饰器模式：WatermarkStrategyWithIdleness 的两个委托

看主干代码就很清楚（删掉了断言和序列化部分）：

static class WatermarkStrategyWithIdleness<T> implements WatermarkStrategy<T> { private final WatermarkStrategy<T> baseStrategy; private final Duration idlenessTimeout; @Override public TimestampAssigner<T> createTimestampAssigner(TimestampAssignerSupplier.Context context) { return baseStrategy.createTimestampAssigner(context); } @Override public WatermarkGenerator<T> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) { return new WatermarksWithIdleness<>( baseStrategy.createWatermarkGenerator(context), idlenessTimeout); } }

两个方法各干一件事：时间戳分配器原样透传，水位生成器被包了一层WatermarksWithIdleness。这意味着withIdleness的调用位置很灵活——放在withTimestampAssigner前面还是后面都行，因为它包装的是“生成器”这一侧，跟时间戳分配互不干扰。

顺带说一个容易被忽略的点：withIdleness是可叠加的，但没必要。你调两次，就会得到两层WatermarksWithIdleness，外层按自己的超时判定，内层也按自己的超时判定，最终生效的是两者中更严格的那个。这种写法除了给自己添堵没别的好处。

2.2 IdlenessTimer 用处理时间而不是事件时间

WatermarksWithIdleness本体非常短：

public class WatermarksWithIdleness<T> implements WatermarkGenerator<T> { private final WatermarkGenerator<T> watermarks; private final IdlenessTimer idlenessTimer; @Override public void onEvent(T event, long eventTimestamp, WatermarkOutput output) { idlenessTimer.activity(); watermarks.onEvent(event, eventTimestamp, output); } @Override public void onPeriodicEmit(WatermarkOutput output) { if (idlenessTimer.checkIfIdle()) { output.markIdle(); } else { watermarks.onPeriodicEmit(output); } } }

这里有个非常关键的设计决定，值得停下来想三秒：空闲判定用的是处理时间，不是事件时间。

原因很直接。如果用事件时间判定，那“空闲”就意味着“没有新的事件时间推进”，而水位本身就是事件时间的函数，你会陷入循环论证。更实际的原因是：一个数据源有没有数据，是一个物理事实，跟数据里带的时间戳无关。上游停摆了，机器还在跑，墙上的钟还在走，用System.nanoTime()判定最直接。

IdlenessTimer的核心逻辑大概是这样：

static class IdlenessTimer { private long counter; private long startOfInactivityNanos; private final long maxIdleTimeNanos; public void activity() { counter++; } public boolean checkIfIdle() { final long currentTime = System.nanoTime(); if (startOfInactivityNanos == 0) { startOfInactivityNanos = currentTime; return false; } if (counter > 0) { counter = 0; startOfInactivityNanos = currentTime; return false; } return currentTime - startOfInactivityNanos >= maxIdleTimeNanos; } }

读这段代码要注意三个细节。第一，startOfInactivityNanos == 0是个哨兵判断，第一次调用只记录起点、不下结论，所以真正生效的等待时间比配置的超时多一个检测周期，可以忽略。第二，判活用的是counter，只要onEvent被调用过就自增，checkIfIdle里读到非零就重置起点，所以活跃期的检测是“自上次检查以来有没有来过数据”，粒度就是检测周期本身。第三，用System.nanoTime()而不是System.currentTimeMillis()，是为了不受系统时钟回拨影响——这点在容器环境里尤其重要。

withIdleness构造时会校验超时必须为正数，传Duration.ZERO会直接抛异常。这个校验挺有必要，因为零超时意味着“每两个检测周期之间没数据就算空闲”，在正常抖动的流量下会把数据源反复标成 idle，行为非常诡异。

2.3 onPeriodicEmit 是空闲检测的“心跳”

上面那段代码里，markIdle只在onPeriodicEmit里被调用。这句话的分量比它看起来重得多：空闲检测完全依赖周期性的onPeriodicEmit调用，如果这个调用不发生，withIdleness就是一段死代码。

onPeriodicEmit由算子的处理时间服务周期性触发，间隔取自pipeline.auto-watermark-interval，默认 200 毫秒。也就是说，默认情况下每 200 毫秒会检查一次“这条通道最近有没有数据”。

这个链条一旦断掉，后果就是静默失效。最常见的断法是把自动水位间隔设成 0：

env.getConfig().setAutoWatermarkInterval(0);

这么写的初衷通常是“我不想让 Flink 自动发水位，我要完全自己控制”，但副作用是把withIdleness一起关掉了。我见过一个任务这么配了之后，加了withIdleness却毫无效果，排查了大半天，最后发现是这行配置。所以如果你既要关自动水位、又要空闲检测，只能自己在onEvent里做判定并显式调用markIdle，这部分在第 7 章会给写法。

3. markIdle 之后，水位是怎么绕开空闲通道的

output.markIdle()只是起点。这个调用之后，信号要经过算子输出侧、网络传输、下游算子输入侧三个环节，才能真正影响到最小水位的计算。这一段链路是理解“为什么有时候 markIdle 了还是不生效”的关键。

3.1 WatermarkOutputMultiplexer 的合并逻辑

一个算子可能有多个下游输出（侧输出、分流），但它对外只有一个WatermarkOutput。WatermarkOutputMultiplexer就是干这个聚合的：它给每个下游输出维护一份独立的 watermark 状态，算子内部任何一个输出发水位，都通过它转发并合并。

当某个输出调用markIdle时，multiplexer 会把这份输出标记成空闲，之后计算合并水位时把它排除在外。如果所有输出都空闲了，multiplexer 就停止向下游发新的水位——注意是“停止发”，不是“发一个最大值”。

在 1.14 之后的 Source 实现里，这个组件的位置变得更重要了，因为每个 split（Kafka 的一个分区、文件的一个分片）都有独立的输出，multiplexer 要负责把它们的水位合并成算子级别的水位。这个变化在第 4 章细说。

3.2 StatusWatermarkValve 里的通道状态数组

信号穿过网络到达下游算子后，落到StatusWatermarkValve手里。它做的事可以概括成三步：

1. 收到WatermarkStatus.IDLE时，把对应 channel 的对齐标记置为 false；
2. 收到WatermarkStatus.ACTIVE或者一条新水位时，把对齐标记置回 true，并更新该通道的水位值；
3. 重新在所有对齐通道里取最小值，如果这个最小值比上次发出去的水位大，就往下游发。

有一个行为容易被误解：当所有通道都变成 idle 时，valve 不会发出任何新水位。下游算子会保持在上一次收到的那条水位上，停在那里。所以你不会看到水位突然跳到Long.MAX_VALUE，而是看到它“冻住”了。Web UI 上显示No Watermark (Long.MAX_VALUE)是界面层对 idle 状态的展示方式，跟实际水位值不是一回事，别被这个显示误导。

3.3 从 idle 回到 active 的那一瞬间

这一段是最容易出问题的。假设某条通道空闲了 10 分钟，期间全局水位已经推进到了T+600s，现在这条通道突然来了一条时间戳是T+1s的数据。

会发生什么？这条通道的对齐标记被置回 true，它的水位重新参与最小值计算。但它的水位还停在T+1s附近，于是全局水位立刻被拽回到T+1s—— 但水位是单调的，不能回退，所以这次“拽回”会被直接忽略，下游水位保持T+600s。

结果就是：这条通道后续发出的所有时间戳小于T+600s的数据，在下游看来都是迟到数据。窗口如果配了allowedLateness，还能救回来一批；没配的话，直接进侧输出或者被丢弃。

这不是withIdleness的缺陷，而是“长期空闲后突然恢复”这个场景的固有代价。能做的是两件事：把空闲超时设得比业务上“正常情况下最长的静默期”更长一些，以及在窗口上配置合理的allowedLateness做缓冲。这两条我会在第 6 章展开。

4. 版本迁移带来的粒度变化：从 subtask 到 split

这一章可能是全文最有实用价值的部分，因为它解释了一个很多人踩过但没想明白的现象：为什么同样的代码，升级 Flink 版本之后withIdleness的效果变了。

4.1 1.14 之后空闲检测下沉到 SourceReader

在 1.13 及之前的版本里，你去看WatermarkStrategy.withIdleness的实现，会找到WatermarksWithIdleness这个包装类，就是第 2 章拆的那个。在这个模型下，空闲检测发生在水位生成器这一层，而水位生成器是跟着算子走的，一个 source 子任务只有一个。

1.14 之后，随着 Source 接口（FLIP-27）的推广，这个结构变了。改进单 FLINK-23683 的核心就是把空闲检测从水位生成器里拆出来，下沉到 Source 读取层。你去翻新版本的源码，会发现WatermarksWithIdleness已经不在了，取而代之的是SourceReaderBase在给每个 split 创建输出对象时，一并创建了一个空闲管理器。

伪代码大致是这样：

// 每个 split 一份输出，输出里带着自己的空闲管理器 class SourceOutputWithWatermarks<T> extends SourceOutput<T> { private final IdlenessManager idlenessManager; @Override public void collect(T record) { idlenessManager.onActivity(); // 有数据就报活 // ... 正常往下游发 } public void emitPeriodicWatermark() { if (idlenessManager.isIdle()) { sourceOutput.markIdle(); // 没数据就报空闲 } else { watermarkGenerator.onPeriodicEmit(sourceOutput); } } }

对外的 API 完全没变，你写的还是.withIdleness(Duration.ofSeconds(30))，但内部的判定位置和判定粒度都换了。这个“API 不变、语义微调”的升级最容易出事。

4.2 一个 subtask 消费多个分区时，旧版会漏判

粒度变化带来的实际差异，用一个具体配置来说明最清楚：一个 Kafka 数据源，4 个并行子任务，每个子任务消费 5 个分区，总共 20 个分区。

旧版（1.13 及之前）：空闲检测是 subtask 粒度的。这个子任务的水位取它消费的 5 个分区的最小值。只要这 5 个分区里有任意一个有数据，onEvent就会被调用，counter就自增，这个子任务就永远不算空闲。于是另外 4 个早就没有数据的分区，会一直把它们的分区水位钉在那里，拖住整个子任务的水位。这时候withIdleness完全无能为力——它看不见分区级别的事情。

新版（1.14 及之后）：空闲检测是 split 粒度的。每个分区有自己的输出对象和空闲管理器，某个分区 30 秒没数据，就单独把这个 split 标记成空闲，退出最小值比较。剩下有数据的分区照常推进水位。

这个差异在分区数量远大于并行度、并且流量分布不均的场景下非常明显。我遇到过一次升级之后“窗口突然开始按时触发了”，业务方以为是升级修了 bug，其实是粒度变化把原来被拖住的水位放开了。理解这一点，你就能判断升级前后行为差异是不是预期内的。

4.3 版本行为对照表

把关键差异整理成一张表，排查时对照着看：

表格最后两行值得多说一句。新版把检测下沉到 Source 层之后，对于那些“在中间算子重新生成水位”的写法（比如 union 之后重新assignTimestampsAndWatermarks），withIdleness的行为就跟旧版不一样了。如果你的场景是“两条流 union，其中一条可能长时间没数据”，比较稳妥的做法是在 union之前分别给每条流设置水位策略（第 6 章会讲）。

5. 手把手跑一个能看见 idle 生效的 Demo

理论讲完，跑一个最小可复现的例子。这个 Demo 我在本地和测试集群上都跑过，用最少的代码把“水位卡住 → 加 withIdleness → 水位恢复”这个过程完整走一遍。

5.1 构造一个“半死不活”的数据源

核心是一个并行度 2 的 source：子任务 0 持续发数据，子任务 1 发几条之后彻底沉默。为了简单，这里用RichParallelSourceFunction，生产环境当然要用新 Source API，但演示效果一致。

public class HalfDeadSource extends RichParallelSourceFunction<Event> { private volatile boolean running = true; @Override public void run(SourceContext<Event> ctx) throws Exception { int index = getRuntimeContext().getIndexOfThisSubtask(); if (index == 1) { // 子任务 1：发 3 条就彻底沉默，模拟空闲分区 for (int i = 0; i < 3; i++) { synchronized (ctx.getCheckpointLock()) { ctx.collect(new Event("k1", 1000L * i, 1L)); } Thread.sleep(100); } while (running) { Thread.sleep(500); } return; } // 子任务 0：每 100ms 一条，事件时间步进 1 秒 long ts = 0; while (running) { synchronized (ctx.getCheckpointLock()) { ctx.collect(new Event("k1", ts, 1L)); } ts += 1000; Thread.sleep(100); } } @Override public void cancel() { running = false; } }

Event就是个普通 POJO，三个字段：key、事件时间、值。注意ctx.collect外面套了getCheckpointLock()，这是老 API 的规范写法，不加会在 checkpoint 时出现数据与快照不一致。

然后是主逻辑，先跑不加withIdleness的版本：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); env.enableCheckpointing(10_000); DataStream<Event> stream = env .addSource(new HalfDeadSource()) .setParallelism(2) .assignTimestampsAndWatermarks( WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(5)) .withTimestampAssigner((e, ts) -> e.getEventTime()) ); stream.keyBy(Event::getKey) .window(TumblingEventTimeWindows.of(Time.seconds(5))) .process(new WindowResult()) .print();

窗口 5 秒，乱序容忍 5 秒，正常情况下子任务 0 的数据足够让窗口按秒级节奏触发。但加上子任务 1 这条“静止的腿”，窗口会一条都不出。

5.2 加上 withIdleness 前后的对比

把策略换成带空闲检测的版本，其余不变：

WatermarkStrategy<Event> strategy = WatermarkStrategy .<Event>forBoundedOutOfOrderness(Duration.ofSeconds(5)) .withTimestampAssigner((e, ts) -> e.getEventTime()) .withIdleness(Duration.ofSeconds(10));

跑起来之后，观察到的现象是：

• 前 10 秒：窗口依然一个都不出，因为子任务 1 还没被判定为空闲；
• 大约第 10 到 11 秒之间：子任务 1 被标记 idle，全局水位开始恢复推进；
• 之后：窗口按每 5 秒一个的节奏稳定输出。

这个“前 10 秒没输出”不是 bug，是空闲超时的必然代价。超时越短，恢复越快，但误判风险越高；超时越长，越稳，但恢复越慢。这个取舍在第 6 章会展开讲。

5.3 用三个位置验证 idle 真的发生了

光看窗口有没有输出不够，万一它是别的原因触发的呢？我一般会在这三个位置看：

位置一：窗口后的算子打印当前水位。加一个ProcessFunction，在onTimer或者processElement里读ctx.timerService().currentWatermark()。加withIdleness之前它一直是Long.MIN_VALUE + 1，加之后会从某个值开始往上跳。

stream.keyBy(Event::getKey) .process(new KeyedProcessFunction<String, Event, String>() { @Override public void processElement(Event e, Context ctx, Collector<String> out) { long wm = ctx.timerService().currentWatermark(); out.collect("wm=" + wm + " eventTs=" + e.getEventTime()); } @Override public void onTimer(long ts, OnTimerContext ctx, Collector<String> out) { } });

位置二：Web UI 的 Watermarks 面板。看每个 task 的 Last Received Watermark 和 idle 标记。这里要提醒一句：面板上会显示通道的对齐状态，某条通道是 idle 时，它的水位显示可能是个特殊值，不等同于实际水位，别拿它当准数。

位置三：自带指标。TaskManager 的指标里有一组水位相关的 gauge，能看到每个 task 当前的水位值。如果你在容器里跑、Web UI 不方便开，抓这个指标比日志靠谱。

跑通这个 Demo 之后，建议再做一次反向验证：把withIdleness删掉，确认窗口真的不出；把超时改成 1 毫秒，确认窗口立刻就开始出。这两次对照做完，你对这个参数的手感就建立了。

6. 五个真踩过的坑和对应的处理方式

前面讲原理是为了这一章服务。这些都是我在实际项目里遇到过、并且花时间定位过的，不是从文档里抄的。

6.1 auto-watermark-interval 为 0 时 withIdleness 直接失效

第一章提过一次，这里展开说