1. 先还原现场watermark 为什么会被一个“没数据的分区”拖住1.1 一个典型的凌晨告警先说一个我自己踩过的场景。某天凌晨值班同学发来消息一个按事件时间开窗的 Flink 作业消费速率正常、checkpoint 正常、CPU 也不高但窗口已经四十分钟没有输出Web UI 上某个 source 子任务的 watermark 一直停在同一个时间点。更迷惑的是这个作业前一天跑得好好的流量也没降。把 source 的并行度和上游分区情况拉出来一看问题很清楚了上游一共有 16 个分区其中 3 个分区在凌晨基本没有数据而作业的 watermark 取的是所有输入通道的最小值。只要这 3 个分区不产生新事件它们对应的 watermark 就不动全局 watermark 就被按在原地后面的窗口自然一个都触发不了。这类问题在多分区消息队列、分库分表 binlog、多日志文件采集、多路 HTTP 拉取等场景里非常常见。它的迷惑性在于作业没挂数据也在处理只是事件时间停住了。如果你只盯着吞吐和异常日志很容易漏掉 watermark 这条线。WatermarkStrategy.withIdleness()就是 Flink 给这类“空闲数据源”准备的标准解法。它的作用不是跳过数据也不是删除分区而是在某个输入通道一段时间没有数据时把它从全局 watermark 的最小值计算里暂时摘出去等数据回来再重新纳入。下面我按“现象—原理—源码—配置—踩坑”的顺序把这套机制拆开讲一遍。1.2 木桶效应watermark 取的是所有通道的最小值理解 idle 机制先要理解 watermark 的合并规则。Flink 里一个算子的输入可能有多个 channelsource 的多个分区、上游多个算子的输出、union 之后的多个输入流。每个 channel 都有自己的 watermark算子对外发出的 watermark 取所有活跃 channel的最小值。这个设计很好理解只有所有通道都确认“时间 t 之前的数据到齐了”算子才敢说“我的事件时间已经到 t 了”。问题就出在“所有通道”这四个字上。假设通道 A 每秒来 100 条数据watermark 飞快往前推通道 B 半小时才来一条watermark 半天不动。那全局 watermark 就只能跟着 B 走。窗口结束时间只要比 B 的 watermark 大就不会触发。这就是典型的木桶效应最短的那块板决定水位。注意这里说的“没数据”不等于“没 watermark”。有些 source 即使没有新事件也会周期性地把当前 watermark 再发一遍但值不变对全局最小值没有贡献。真正能解决问题的是让下游知道“这个通道现在没有事件时间进展别等它”也就是 idle 标记。1.3 idle 机制到底做了什么withIdleness(Duration idleTimeout)的语义可以一句话概括某个输入通道超过idleTimeout没有收到事件就把它标记为 idle后续计算全局 watermark 时忽略它一旦有新事件到达再自动恢复为 active。这里有几个容易混淆的点我逐个说清楚idle 判定用的是处理时间不是事件时间。它看的是“我多久没收到这条通道的数据了”跟数据里带的时间戳无关。所以即使一条数据的事件时间是十年前的只要它刚到达通道就算活跃。idle 不是丢弃数据。标记 idle 之后通道仍然可以接收数据新数据到达后通道会重新参与 watermark 计算。idle 是逐通道的。source 的每个分区、每个 split、每个 input channel 都有自己的状态不是整个 source 一起 idle。idle 只影响 watermark 合并不影响窗口逻辑本身。窗口什么时候触发仍然由事件时间和 watermark 决定只是 watermark 现在能往前走了。还有一个很重要的边界如果所有通道都 idle 了Flink 会把 watermark 推到Long.MAX_VALUE也就是最大值。这个行为本身是为了让下游知道“暂时不会有数据了”但它会带来一个副作用所有事件时间窗口会立刻触发并清理。等数据再回来就可能被当成迟到数据。这个坑我在第 4 节会展开讲。2. 顺着源码走一遍withIdleness 是怎么包上去的2.1 WatermarkStrategy.withIdleness 的入口与参数校验WatermarkStrategy是一个接口里面有两个核心方法createTimestampAssigner负责从数据里提取事件时间createWatermarkGenerator负责生成 watermark。withIdleness是接口上的一个 default 方法返回的是一个包装后的新策略。简化后的源码大致是这样default WatermarkStrategyT withIdleness(Duration idleTimeout) { checkNotNull(idleTimeout, idleTimeout); checkArgument( !(idleTimeout.isZero() || idleTimeout.isNegative()), idleTimeout must be greater than zero); return new WatermarkStrategyT() { Override public WatermarkGeneratorT createWatermarkGenerator( WatermarkGeneratorSupplier.Context context) { return new WatermarksWithIdleness( WatermarkStrategy.this.createWatermarkGenerator(context), idleTimeout); } Override public TimestampAssignerT createTimestampAssigner( TimestampAssignerSupplier.Context context) { return WatermarkStrategy.this.createTimestampAssigner(context); } }; }这段代码有两个细节值得注意。第一idleTimeout必须严格大于零。你传Duration.ZERO或者负数作业启动阶段就会直接抛参数异常。这个校验很合理零超时意味着“任何一次没有事件的瞬间都算空闲”加水印的节奏会被打得稀碎。第二withIdleness用的是装饰器模式。它没有修改原来的 timestamp assigner也没有改原来 generator 的内部逻辑只是在外层套了一个WatermarksWithIdleness。这意味着你可以把它加在任意WatermarkStrategy后面比如forBoundedOutOfOrderness、forMonotonousTimestamps甚至你自己实现的策略。实际写代码时withIdleness和withTimestampAssigner的调用顺序不影响最终效果因为两者都是通过匿名策略逐层委托的。但从可读性角度我一般建议先写时间戳提取再写空闲超时WatermarkStrategyEvent strategy WatermarkStrategy.EventforBoundedOutOfOrderness(Duration.ofSeconds(10)) .withTimestampAssigner((event, recordTimestamp) - event.getEventTime()) .withIdleness(Duration.ofMinutes(2));这样读起来像是“先定义事件时间怎么来再定义没数据时怎么办”逻辑顺序更自然。2.2 WatermarksWithIdleness事件、周期发射与处理时间定时器WatermarksWithIdleness是org.apache.flink.api.common.eventtime包下的类它同时实现了WatermarkGenerator和ProcessingTimeService.ProcessingTimeCallback。前者说明它要接管 watermark 生成后者说明它能注册处理时间定时器。它的核心字段只有两个被包装的WatermarkGenerator和一个IdlenessTimer。简化后的逻辑如下public class WatermarksWithIdlenessT implements WatermarkGeneratorT, ProcessingTimeService.ProcessingTimeCallback { 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); } } Override public void onProcessingTime(long timestamp) { if (idlenessTimer.checkIfIdle()) { output.markIdle(); } } }这段代码虽然简化了字段和定时器注册但把关键路径说清楚了每来一条事件先“报个到”。idlenessTimer.activity()会刷新最后一次活动时间然后才把事件交给底层 generator 处理。这个顺序很重要保证空闲计时不会被底层逻辑拖慢。周期性发射时先检查是否空闲。如果已经空闲直接调用output.markIdle()不再让底层 generator 发 watermark如果不空闲继续走原来的发射逻辑。处理时间定时器是独立兜底。即使onPeriodicEmit因为某些原因没有及时调用定时器回调也会检查空闲状态并标记。这样空闲检测不完全依赖pipeline.auto-watermark-interval的节奏。这里有一个实际使用中的关键点output.markIdle()之后底层 generator 并没有被销毁它只是暂时不参与全局 watermark 的最小值计算。等新事件到达onEvent会刷新活动时间下一次emitWatermark或通道状态更新时通道会重新变为活跃。2.3 IdlenessTimer 的判定逻辑IdlenessTimer是WatermarksWithIdleness的内部辅助类逻辑不复杂但有几个设计取舍值得说。它记录的是“最后一次活动的时间点”单位是纳秒用的是System.nanoTime()这类单调时钟而不是墙上时钟。这么做是为了避免系统时间被调整时导致空闲判断错乱。每次activity()被调用就更新这个时间点每次checkIfIdle()被调用就用当前时间减去最后活动时间跟idleTimeout比较。伪代码大概是这样boolean checkIfIdle() { long idleNanos System.nanoTime() - lastActivityNanos; return idleNanos idleTimeoutNanos; }这里有两个实际影响第一检测精度大约是一个idleTimeout。定时器不是每毫秒都在跑它按超时时间注册回调。所以真实空闲时间可能比配置值多出一点但不会少太多。对大多数业务来说这个误差可以接受。第二空闲判定跟数据里的事件时间完全解耦。一条事件时间很旧的数据到达也会刷新活动时间一条事件时间很新的数据没到达也不会阻止空闲判定。这一点在排查问题时特别容易搞混。有人看到“watermark 明明已经推到很前面了怎么还被标记 idle”其实是因为那条通道确实很久没有新事件了跟 watermark 的值没关系。另外IdlenessTimer的activity()只在onEvent里调用不会在onPeriodicEmit里调用。也就是说只有真实数据到达才算“活动”周期性的空发射不算。这个设计是合理的否则一个每秒空发 watermark 的 source 永远不会被判定为空闲idle 机制就失效了。2.4 下游 StatusWatermarkValve 如何排除 idle 通道source 端调用markIdle()只是第一步真正让全局 watermark 往前走的是下游算子的合并逻辑。Flink 在算子输入侧用StatusWatermarkValve来管理多个输入通道的 watermark 和状态。它内部为每个 channel 维护一个ChannelStatus里面至少有两个信息当前 watermark 值和是否空闲。每当某个通道发来新的 watermark 或状态更新它都会重新计算一次“所有活跃通道的最小 watermark”然后决定是否向下游发射。关键逻辑可以概括为// 收到普通 watermark更新该通道的值并置为活跃 channel.watermark watermark.getTimestamp(); channel.isIdle false; // 收到 idle 标记把该通道置为空闲 channel.isIdle true; // 重新计算只遍历活跃通道取最小值 long min Long.MAX_VALUE; boolean hasActive false; for (ChannelStatus channel : channels) { if (!channel.isIdle) { min Math.min(min, channel.watermark); hasActive true; } } if (hasActive) { output.emitWatermark(new Watermark(min)); } else { output.emitWatermark(new Watermark(Long.MAX_VALUE)); }这段伪代码解释了三个现象某个通道 idle 后它的旧 watermark 不再参与最小值计算全局 watermark 可以跳到其他活跃通道的最小值。所有通道都 idle 时输出Long.MAX_VALUE下游会认为“事件时间到顶了”。一个 idle 通道重新收到数据后会先变成活跃但它的 watermark 可能还停留在旧值全局 watermark 会被它重新拉回来一点但不会回退已经发出的 watermark只是暂时停止推进等它追上来。还有一个容易忽略的组件是WatermarkOutputMultiplexer。在 source 内部一个 subtask 可能同时读多个分区每个分区都有自己的 watermark generator这些 generator 的输出会通过 multiplexer 合并成一个对外 watermark。合并规则跟StatusWatermarkValve类似活跃输出取最小值全部空闲则输出最大值。所以withIdleness在 source 内部和下游算子两侧是配套工作的不是只在一个地方生效。3. 落地配置代码写法、参数估算与窗口配合3.1 三种常见接入方式的写法实际项目里withIdleness一般出现在三个位置source 的 watermark 策略、assignTimestampsAndWatermarks的显式分配、以及自定义 generator 的内部。写法都不复杂但要注意别把顺序和位置搞错。第一种source 端直接指定。现在主流的 source 实现都支持在构造时传入WatermarkStrategy这是最推荐的方式因为 watermark 在 source 端生成最早、最直接DataStreamEvent stream env.fromSource( source, WatermarkStrategy.EventforBoundedOutOfOrderness(Duration.ofSeconds(10)) .withTimestampAssigner((event, ts) - event.getEventTime()) .withIdleness(Duration.ofMinutes(2)), event-source);第二种在数据流上显式分配。适用于 source 没有内置 watermark 策略或者你想在 source 之后统一处理时间戳的场景DataStreamEvent stream env.fromSource(source, WatermarkStrategy.noWatermarks(), event-source) .assignTimestampsAndWatermarks( WatermarkStrategy.EventforBoundedOutOfOrderness(Duration.ofSeconds(10)) .withTimestampAssigner((event, ts) - event.getEventTime()) .withIdleness(Duration.ofMinutes(2)));这里要注意如果 source 端已经用了noWatermarks()后面再用assignTimestampsAndWatermarks是没问题的但如果 source 端已经生成过 watermark后面再分配一次容易出现“两套 watermark 打架”的情况。我的建议是watermark 只在一个地方生成别两头都配。第三种自定义WatermarkGenerator内部主动调用markIdle()。适合你有特殊空闲判定逻辑的场景比如“连续 N 个周期没有达到某个业务条件就标记空闲”public class BusinessIdleGenerator implements WatermarkGeneratorEvent { private final long idleTimeoutMillis; private long lastEventTime System.currentTimeMillis(); private long maxTimestamp Long.MIN_VALUE; public BusinessIdleGenerator(long idleTimeoutMillis) { this.idleTimeoutMillis idleTimeoutMillis; } Override public void onEvent(Event event, long eventTimestamp, WatermarkOutput output) { lastEventTime System.currentTimeMillis(); maxTimestamp Math.max(maxTimestamp, eventTimestamp); output.markActive(); } Override public void onPeriodicEmit(WatermarkOutput output) { long now System.currentTimeMillis(); if (now - lastEventTime idleTimeoutMillis) { output.markIdle(); } else { output.emitWatermark(new Watermark(maxTimestamp - 5_000)); } } }这段代码里markActive()和markIdle()是成对出现的。实际使用中底层通道在收到新 watermark 时通常会自动恢复活跃但显式调用markActive()能让意图更清晰也避免某些自定义输出实现的状态不一致。3.2 idleTimeout 到底设多少一个可落地的估算方法这是被问得最多的问题也是配置错误最集中的地方。很多人凭感觉写个Duration.ofSeconds(10)结果正常抖动被误判成空闲窗口结果忽多忽少。我的经验是idleTimeout要同时满足两个条件大于正常消息间隔的波动上限小于业务能接受的端到端延迟上限。先看下限。假设某个分区平时每 30 秒来一条数据但偶尔会因为上游批量写入、网络抖动、定时任务等原因出现 2 到 3 分钟的空档。那把idleTimeout设成 30 秒就很容易误判。一个比较稳妥的起点是idleTimeout ≈ max(3 × P99 消息间隔, 2 × 上游批量写入周期)P99 消息间隔可以从监控里取也可以先粗估。比如 P99 是 20 秒上游每 5 分钟批量刷一次那idleTimeout至少应该是 10 分钟否则每次批量间隔都会被当成空闲。再看上限。idleTimeout直接加到端到端延迟上一个通道空闲后最多要等idleTimeout才被标记全局 watermark 才能往前走。如果业务要求窗口结果在事件发生后 1 分钟内输出那idleTimeout就不能超过 1 分钟否则延迟指标一定不达标。这时候要么调小超时、接受一定的误判风险要么把空闲分区拆出去单独处理。给一个粗略的参考表实际值一定要结合自己的监控调数据特征典型消息间隔建议 idleTimeout 起点说明高频稳定流10 到 100 毫秒10 到 30 秒不需要太小避免网络抖动误判中频业务流1 到 5 秒30 到 60 秒大多数多分区场景的常用区间低频日志流1 到 5 分钟10 到 15 分钟必须覆盖正常的采集空档定时批量流按小时或按天大于批量周期否则每次批量间隔都会触发 idle还有一个组合关系要记住idleTimeout和forBoundedOutOfOrderness的乱序容忍时间是两个独立参数。前者管“多久没数据算空闲”后者管“事件时间允许乱序多少”。不要因为调大了idleTimeout就顺便调大乱序时间也不要指望乱序时间能解决 watermark 卡住的问题。3.3 和 allowedLateness、sideOutputLateData 的配合withIdleness让 watermark 能往前走副作用是窗口可能比预期更早触发原本还能赶上窗口的数据现在可能变成迟到数据。所以只要用了 idle我一般都建议把迟到数据的兜底配上。stream.keyBy(Event::getKey) .window(TumblingEventTimeWindows.of(Time.minutes(5))) .allowedLateness(Time.minutes(2)) .sideOutputLateData(lateTag) .process(new MyWindowFunction());allowedLateness的作用是窗口触发后不立刻销毁继续保留一段时间期间到达的迟到数据还能触发增量计算。sideOutputLateData则是把超过容忍期的数据分流到侧输出避免静默丢弃。这两个配置和idleTimeout的关系可以这样理解idleTimeout决定“多久不管这个通道”allowedLateness决定“窗口关掉之后还等多久”。如果某个低频分区经常在 idle 之后才来数据那allowedLateness至少要覆盖“idle 判定时间 数据回追时间”否则侧输出会明显增多。我一般会先看侧输出流的数据量和时间分布再反推参数。如果侧输出里的数据事件时间集中在窗口结束后 1 分钟内说明idleTimeout偏小或者allowedLateness偏小如果侧输出数据很少说明当前组合基本合适。3.4 多源 union/connect 与分区倾斜场景除了 source 内部的多分区withIdleness还经常用在多流合并的场景。比如一条主流量和一条配置流做connect配置流可能几小时才更新一次。如果不做空闲处理配置流的 watermark 会把整个算子的 watermark 拖住。这种场景要注意withIdleness要加在每个可能空闲的输入流上而不是只加在主流程上。connect之后两条流各自保留自己的 watermark 状态只配一边是没用的。DataStreamEvent main env.fromSource(mainSource, WatermarkStrategy.EventforBoundedOutOfOrderness(Duration.ofSeconds(5)) .withTimestampAssigner((e, ts) - e.getEventTime()) .withIdleness(Duration.ofMinutes(1)), main-source); DataStreamConfig config env.fromSource(configSource, WatermarkStrategy.ConfigforBoundedOutOfOrderness(Duration.ofSeconds(5)) .withTimestampAssigner((c, ts) - c.getEventTime()) .withIdleness(Duration.ofMinutes(10)), config-source); main.connect(config).process(new MyCoProcessFunction());这里两个流的idleTimeout可以不一样。主流量密集1 分钟足够配置流稀疏10 分钟更稳。不要图省事统一设一个值稀疏流的误判成本更高。分区倾斜是另一个常见诱因。比如 16 个分区里3 个分区承担了 90% 的数据剩下 13 个分区几乎没数据。这时候即使配了 idle也要观察 watermark 的推进曲线如果它呈现“走一段停一段”的锯齿状说明空闲判定和恢复在反复切换。可以考虑从数据分布上做重分区或者把低频分区单独拆一个作业处理。withIdleness是运行时兜底不是数据倾斜的根治方案。4. 踩坑记录withIdleness 不是万能药4.1 全 idle 会把 watermark 推到最大值这是最需要警惕的行为。前面说过当所有输入通道都 idle 时StatusWatermarkValve会输出Long.MAX_VALUE。下游收到这个 watermark会认为事件时间已经到顶所有事件时间窗口立即触发并清理。如果你的业务在凌晨会有整条流完全静默的时段而idleTimeout又设得比较小就会出现凌晨两点所有分区都没数据watermark 跳到最大值窗口全部触发凌晨三点数据恢复新数据的事件时间远小于最大值全部被当成迟到数据要么进侧输出要么直接丢弃。我见过一个更隐蔽的版本作业白天正常凌晨窗口集中输出一大批空结果第二天早上业务方发现报表全是零。排查了半天才发现是全 idle 导致的。应对方式有三种按推荐程度排序把idleTimeout设得比最长静默时段还长。如果业务每天凌晨静默 2 小时idleTimeout至少设 2 小时以上或者干脆不在这个时段依赖 idle。对全静默场景做特殊处理。比如在 source 端加一个低频心跳保证至少有一个通道始终活跃或者用处理时间窗口替代事件时间窗口。接受窗口提前触发但把allowedLateness放大。这只适合窗口状态不大、迟到数据可接受的场景不是通用方案。4.2 低频分区被误判数据回来变成迟到第二个坑是“假空闲”。某个分区其实有数据只是间隔比较长idleTimeout设小了它被标记 idle等数据回来时全局 watermark 已经跑到前面去了这条数据就成了迟到数据。这种问题的典型表现是侧输出流的数据量随时间缓慢增长而且这些数据的事件时间总是比当前 watermark 小一截。排查时可以把侧输出数据按分区、按事件时间打个分布如果集中在某几个低频分区基本就能确认是误判。处理思路先调大idleTimeout观察侧输出是否下降。如果调大后延迟不达标考虑把这个低频分区拆到独立作业用独立的 watermark 策略处理。配合allowedLateness和侧输出至少保证数据不丢后续可以离线补算。这里有个经验不要用平均消息间隔来估idleTimeout要用 P99 甚至 P999。平均值会被高频数据拉低而触发误判的往往是那些长尾空档。4.3 恢复后状态不继承别指望立刻生效withIdleness的空闲状态和定时器是基于处理时间的运行时状态不会随着 checkpoint 保存。作业从 checkpoint 恢复后所有通道会重新开始计时需要再经过一个idleTimeout才可能重新标记空闲。这意味着如果作业因为故障重启而某个分区本来就没数据重启后的前idleTimeout时间内watermark 仍然可能被这个分区拖住。如果你把idleTimeout设得很大比如 30 分钟那重启后可能要等半小时才能恢复正常的 watermark 推进。排查时如果看到“重启后 watermark 卡了一段时间又自己好了”不一定是灵异事件很可能就是空闲计时重新开始了。对这个敏感的场景可以把idleTimeout控制在分钟级或者接受重启后的短暂延迟。4.4 排查清单与常用手段下面这张表是我自己排查 watermark 卡住问题时常用的顺序从现象直接对应到检查点现象优先检查常见原因处理方式某个 source 子任务 watermark 不动Web UI 的 Watermarks 页面该子任务对应分区无数据配withIdleness所有子任务 watermark 都不动pipeline.auto-watermark-interval周期发射被设为 0 或异常恢复默认 200ms 或合理值配了 idle 仍不触发窗口是否所有通道都 idle全 idle 输出最大值窗口提前触发调大idleTimeout检查静默时段窗口结果变少、侧输出变多侧输出数据分布idle 误判低频数据被当迟到调大idleTimeout加allowedLateness重启后一段时间又卡重启时间与idleTimeout空闲状态不持久化重新计时接受短暂延迟或调小超时单个分区持续倾斜分区流量分布数据分布不均空闲与活跃反复切换重分区或拆分作业除了看 UI我还会关注两个指标算子级的当前输入 watermark以及迟到数据计数。前者能看出 watermark 是“完全不动”还是“走得很慢”后者能看出 idle 是否造成了误判。如果条件允许可以在自定义 generator 里加一个计数器记录markIdle()的调用次数和恢复次数排查反复切换特别有用。5. 版本差异与替代方案5.1 老版本里的 CombinedWatermarkStatusWatermarkStrategy这套抽象是在 Flink 1.11 引入的withIdleness也随之成为标准能力。在更老的版本里watermark 生成用的是AssignerWithPeriodicWatermarks这类接口空闲检测逻辑散落在 source 的 watermark 合并代码里常见的是CombinedWatermarkStatus这样的内部类。它的基本思路和现在的StatusWatermarkValve类似维护每个通道的 watermark 和空闲状态合并时只取活跃通道的最小值。区别在于 API 形态和可配置性。老版本没有统一的WatermarkStrategy.withIdleness升级时需要注意两点原来通过特殊 watermark 值标记空闲的逻辑要迁移到markIdle()调用上。原来在 source 内部硬编码的空闲超时要改成策略上的withIdleness参数。如果你维护的是跨版本的项目建议把 watermark 策略集中在一个工厂方法里不同版本走不同实现避免升级时到处改。5.2 自定义 WatermarkGenerator 主动 markIdle除了withIdleness你完全可以在自定义WatermarkGenerator里主动控制空闲状态。适合以下几种场景空闲判定不是单纯看“有没有数据”而是看业务字段比如“连续 N 条数据都是心跳包”。需要在空闲恢复时做额外处理比如补发一个补偿 watermark。需要对不同分区用不同的空闲策略。自定义实现的关键是理解WatermarkOutput的三个方法emitWatermark发射 watermarkmarkIdle标记空闲markActive恢复活跃。三者要成对使用别只 markIdle 不 markActive否则通道可能一直处于空闲状态。另外自定义 generator