上个月帮一个团队看一个双流合并的任务现象特别典型两个流都在正常消费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. 拆开 withIdlenessWatermarkStrategy 到底包了什么WatermarkStrategy是一个接口同时继承了TimestampAssignerSupplier和WatermarkGeneratorSupplier。它本身是个“工厂的工厂”你给我上下文我给你造一个时间戳分配器和一个水位生成器。withIdleness没有改变这个契约它只是套了一层装饰器。2.1 装饰器模式WatermarkStrategyWithIdleness 的两个委托看主干代码就很清楚删掉了断言和序列化部分static class WatermarkStrategyWithIdlenessT implements WatermarkStrategyT { private final WatermarkStrategyT baseStrategy; private final Duration idlenessTimeout; Override public TimestampAssignerT createTimestampAssigner(TimestampAssignerSupplier.Context context) { return baseStrategy.createTimestampAssigner(context); } Override public WatermarkGeneratorT createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) { return new WatermarksWithIdleness( baseStrategy.createWatermarkGenerator(context), idlenessTimeout); } }两个方法各干一件事时间戳分配器原样透传水位生成器被包了一层WatermarksWithIdleness。这意味着withIdleness的调用位置很灵活——放在withTimestampAssigner前面还是后面都行因为它包装的是“生成器”这一侧跟时间戳分配互不干扰。顺带说一个容易被忽略的点withIdleness是可叠加的但没必要。你调两次就会得到两层WatermarksWithIdleness外层按自己的超时判定内层也按自己的超时判定最终生效的是两者中更严格的那个。这种写法除了给自己添堵没别的好处。2.2 IdlenessTimer 用处理时间而不是事件时间WatermarksWithIdleness本体非常短public class WatermarksWithIdlenessT implements WatermarkGeneratorT { private final WatermarkGeneratorT 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 毫秒会检查一次“这条通道最近有没有数据”。这个链条一旦断掉后果就是静默失效。最常见的断法是把自动水位间隔设成 0env.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 实现里这个组件的位置变得更重要了因为每个 splitKafka 的一个分区、文件的一个分片都有独立的输出multiplexer 要负责把它们的水位合并成算子级别的水位。这个变化在第 4 章细说。3.2 StatusWatermarkValve 里的通道状态数组信号穿过网络到达下游算子后落到StatusWatermarkValve手里。它做的事可以概括成三步收到WatermarkStatus.IDLE时把对应 channel 的对齐标记置为 false收到WatermarkStatus.ACTIVE或者一条新水位时把对齐标记置回 true并更新该通道的水位值重新在所有对齐通道里取最小值如果这个最小值比上次发出去的水位大就往下游发。有一个行为容易被误解当所有通道都变成 idle 时valve 不会发出任何新水位。下游算子会保持在上一次收到的那条水位上停在那里。所以你不会看到水位突然跳到Long.MAX_VALUE而是看到它“冻住”了。Web UI 上显示No Watermark (Long.MAX_VALUE)是界面层对 idle 状态的展示方式跟实际水位值不是一回事别被这个显示误导。3.3 从 idle 回到 active 的那一瞬间这一段是最容易出问题的。假设某条通道空闲了 10 分钟期间全局水位已经推进到了T600s现在这条通道突然来了一条时间戳是T1s的数据。会发生什么这条通道的对齐标记被置回 true它的水位重新参与最小值计算。但它的水位还停在T1s附近于是全局水位立刻被拽回到T1s—— 但水位是单调的不能回退所以这次“拽回”会被直接忽略下游水位保持T600s。结果就是这条通道后续发出的所有时间戳小于T600s的数据在下游看来都是迟到数据。窗口如果配了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 SourceOutputWithWatermarksT extends SourceOutputT { 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 版本行为对照表把关键差异整理成一张表排查时对照着看维度1.13 及之前1.14 及之后实现类WatermarksWithIdlenessSourceOutputWithWatermarks 空闲管理器检测位置水位生成器层Source 读取层判定粒度source 子任务单个 split分区 / 分片判定时钟System.nanoTime()System.nanoTime()依赖的周期调用算子的onPeriodicEmitSourceReader 的周期水位发射中间算子能否生效能assignTimestampsAndWatermarks处生效主要作用于 Source中间算子场景需注意多分区消费场景容易漏判按分片独立判定表格最后两行值得多说一句。新版把检测下沉到 Source 层之后对于那些“在中间算子重新生成水位”的写法比如 union 之后重新assignTimestampsAndWatermarkswithIdleness的行为就跟旧版不一样了。如果你的场景是“两条流 union其中一条可能长时间没数据”比较稳妥的做法是在 union之前分别给每条流设置水位策略第 6 章会讲。5. 手把手跑一个能看见 idle 生效的 Demo理论讲完跑一个最小可复现的例子。这个 Demo 我在本地和测试集群上都跑过用最少的代码把“水位卡住 → 加 withIdleness → 水位恢复”这个过程完整走一遍。5.1 构造一个“半死不活”的数据源核心是一个并行度 2 的 source子任务 0 持续发数据子任务 1 发几条之后彻底沉默。为了简单这里用RichParallelSourceFunction生产环境当然要用新 Source API但演示效果一致。public class HalfDeadSource extends RichParallelSourceFunctionEvent { private volatile boolean running true; Override public void run(SourceContextEvent 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); DataStreamEvent stream env .addSource(new HalfDeadSource()) .setParallelism(2) .assignTimestampsAndWatermarks( WatermarkStrategy.EventforBoundedOutOfOrderness(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 前后的对比把策略换成带空闲检测的版本其余不变WatermarkStrategyEvent strategy WatermarkStrategy .EventforBoundedOutOfOrderness(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 KeyedProcessFunctionString, Event, String() { Override public void processElement(Event e, Context ctx, CollectorString out) { long wm ctx.timerService().currentWatermark(); out.collect(wm wm eventTs e.getEventTime()); } Override public void onTimer(long ts, OnTimerContext ctx, CollectorString 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 直接失效第一章提过一次这里展开说