Flink checkpoint 对齐把反压放大 3 倍：一次 checkpoint 耗时从 800ms 飙到 22s 的根因、unaligned 开关与它的 3 个隐性坑

一、现象与背景

先给数字。我们一条实时风控特征管道：Flink 1.17.2（K8s 部署，6 TaskManager × 8C32G）、Kafka source 32 分区、RocksDB state backend（增量 checkpoint）、状态总量约 42GB、execution.checkpointing.interval = 30s、超时 120s、min pause between checkpoints = 5s。

平时 checkpoint 耗时 P50 在 400~800ms、P99 在 2.5s，一切正常。但每逢上游 Kafka 出现「分区负载不均」（某几个分区消息量是均值的 25 倍，典型的 hot merchant 场景），checkpoint 指标立刻变形：

• checkpoint duration 从 800ms 飙到 22s（P99 超过 40s，几次直接超 120s 超时被 abort）；
• barrier 对齐等待时间（Flink Web UI 的 Alignment Buffer Size 指标）在 keyBy 算子处堆到 1.7GB / subtask；
• 更糟的是反压倒灌：source 的 busyTimeMsPerSecond 从 180ms 涨到 950ms（几乎打满），Kafka consumer lag 在 3 分钟内涨 48 万条；
• 业务侧：特征新鲜度 SLO（P99 < 8s）连续 20 分钟违约。

一个反直觉的观察：没有任何算子的处理逻辑变慢，CPU 和 GC 都正常，唯一的变量是「分区负载不均」。这直接把矛头指向了 barrier 对齐机制本身。

二、原理：对齐为什么是反压放大器

2.1 aligned checkpoint 的等待语义

Flink 的 checkpoint 靠 barrier 随数据流同步传播实现一致性快照。barrier 是插在数据流里的特殊标记，算子收到所有输入通道的同一编号 barrier 后，才触发本地状态快照、再向下游发 barrier。

关键机制在多输入算子（我们这里的 keyBy 有 32 个输入通道）：

• 快的通道（消息少的分区）barrier 先到了，但算子不能先处理快通道 barrier 之后的数据——否则快照的语义就错了（这些数据属于下一个 epoch）；
• 于是快通道的数据被扣在 buffer 里排队，等慢通道的 barrier 追上来；
• 慢通道的数据来自 hot partition，本身处理就慢（RocksDB 写入排队），barrier 自然滞后；
• 滞后的时间越长，快通道被扣住的数据越多 —— 这就是 alignment buffer。

一句话：aligned checkpoint 用「扣住快通道」换「快照一致性」，扣住的数据 = 反压的物理实体。上游 Kafka 的不均匀负载，被这个机制放大成全链路反压，放大系数取决于「快慢通道的速率差 × 等待时长」。

2.2 为什么平时没事、负载不均时爆炸

平时 32 个分区速率接近，barrier 到达时间差 < 50ms，alignment buffer 只有几 MB，忽略不计。 负载不均时，最快分区 25× 最慢分区，barrier 到达时间差可以到 2~5s，期间最快通道每秒还要持续灌 8000 msg/s 进来 —— 每秒都在往 alignment buffer 里堆数据，30s 一个 checkpoint 周期意味着这个 buffer 有 30s 的累积窗口，堆到 GB 级完全合理。

2.3 unaligned checkpoint 的解法

Flink 1.11 引入 unaligned checkpoint（1.13 起对 RocksDB 支持成熟，1.17 已是稳定特性）：

• barrier 不再等对齐，直接「插队」穿过输入 buffer，向后传播；
• 被扣住的 in-flight 数据不再丢弃，而是纳入算子 state 一起快照（存在 unalign 的额外 state 里）；
• 结果：barrier 传播延迟从「等慢通道」变成「~0」，alignment buffer 指标归零；
• 代价：每个 checkpoint 多快照一份 in-flight 数据（增量 checkpoint 下，这部分也走 RocksDB SST 增量，代价可控），checkpoint 的 state size 会涨 5%~15%（取决于 in-flight 数据量）。

这是典型的「用 state 存储换延迟」的取舍，和我们平时用 RocksDB 换 heap state 是同一类账。

三、具体做法（可复现）

3.1 最小改动：开 unaligned + 保留 aligned 兜底

flink-conf.yaml 或作业代码：

# 开启 unaligned checkpoint
execution.checkpointing.unaligned.enabled: true
# 限制 unaligned 额外 state 大小（防止 in-flight 数据爆炸时 state 爆掉）
state.backend.rocksdb.unaligned-checkpoint.max-additional-size: 512mb

# 其余保持不变
execution.checkpointing.interval: 30s
execution.checkpointing.timeout: 120s
execution.checkpointing.min-pause: 5s
execution.checkpointing.mode: EXACTLY_ONCE
state.backend.type: rocksdb
state.backend.incremental: true

代码方式（Java 17 + Flink 1.17）：

CheckpointConfig cfg = env.getCheckpointConfig();
cfg.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
cfg.setInterval(30_000L);
cfg.setUnalignedCheckpointsEnabled(true);   // 关键开关
cfg.setTolerableCheckpointFailureNumber(3); // 允许连续 3 次失败再 failover

3.2 观测指标（判断 unaligned 是否真的生效）

Flink Web UI → Checkpoints 页签，看：

• End-to-End Duration：应回到 < 2s（我们实测从 22s → 1.3s P99）；
• Alignment Buffer Size：应接近 0（unaligned 不产生 alignment buffer）；
• State Size：会涨 5%~15%，这是预期，不是 bug；
• Unaligned State Size（1.17+ 有独立指标）：反映 in-flight 数据被快照的量，峰值 200~400MB 属正常。

Prometheus 侧关注：

flink_taskmanager_job_task_operator_..._checkpointSize
flink_taskmanager_job_task_operator_..._unalignedStateSize
flink_taskmanager_job_task_operator_..._alignmentBufferBytes  # 应恒为 0

3.3 我们最终的参数与效果

| 指标 | 开 unaligned 前 | 开 unaligned 后 | |---|---|---| | checkpoint duration P50 | 800ms | 620ms | | checkpoint duration P99 | 22s（热分区时段） | 1.3s | | checkpoint 超时次数/天 | 4~7 次 | 0 | | alignment buffer 峰值 | 1.7GB | 0 | | source busyTime P99 | 950ms/1000ms | 210ms/1000ms | | state size（增量） | 42GB | 44.8GB（+6.7%） | | Kafka lag 峰值（热分区时段） | 48 万条 | 3.2 万条 | | 特征新鲜度 SLO 违约时长 | 20min/次 | < 2min/次 |

四、踩过的坑与反直觉之处

坑 1：开了 unaligned，反压还在 —— 因为反压根子不在 checkpoint

我们第一次开 unaligned 后，checkpoint 耗时确实降下来了，但 source busy 还是 800ms。排查发现真正的反压根子是下游 RocksDB compaction 抢占 IO（L0 文件数触顶触发 write stall），checkpoint 对齐只是放大器，不是根因。后来把 level0_slowdown_writes_trigger 从默认 21 调到 35、rate_limiter 限到 200MB/s，反压才真正消失。

启示：unaligned checkpoint 治的是「checkpoint 对齐导致的反压」，不是「所有反压」。先用 flink_taskmanager_job_task_operator_backPressuredTimeMsPerSecond 定位反压源头，再决定要不要开 unaligned。

坑 2：unaligned state 在 failover 恢复时的额外开销

一次 TaskManager OOM 后 failover，恢复耗时从平时的 40s 涨到 3m20s —— 因为 unaligned state 里存的是「当时 in-flight 的数据」，恢复时要把这份数据重新回放进算子逻辑。热分区时段 in-flight 数据大，恢复就慢。

启示：如果业务对 failover 恢复时长敏感（比如 RTO < 1min），开 unaligned 前要把 in-flight 数据的上限算清楚，必要时配合 state.backend.rocksdb.unaligned-checkpoint.max-additional-size 限住。

坑 3：unaligned + Changelog State Backend 不兼容

我们后来想把 checkpoint 换成 Changelog State Backend（增量恢复更快），发现 Changelog State Backend 不支持 unaligned checkpoint（Flink 1.17 文档明确写了）。这是个隐性约束，选型前一定要确认。

五、结论与可复用清单

决策树：什么时候开 unaligned checkpoint

1. checkpoint 耗时 P99 > 5s 且 alignment buffer 指标 > 100MB → 开 unaligned，收益明确；
2. checkpoint 耗时正常，反压来自下游处理慢 → 开 unaligned 没用，先治下游；
3. failover RTO 要求 < 1min 且 in-flight 数据大 → 慎用 unaligned，或限住 max-additional-size；
4. 用 Changelog State Backend → 不能开 unaligned（1.17 限制）。

可复用清单

• [ ] 开 unaligned 前，先记录 baseline：checkpoint P50/P99、alignment buffer 峰值、state size；
• [ ] 开 unaligned 后，验证 alignment buffer 归零、checkpoint P99 下降、state size 涨幅 < 20%；
• [ ] 加 tolerableCheckpointFailureNumber（建议 3），避免单次 checkpoint 失败就 failover；
• [ ] 监控 unalignedStateSize，超过 512MB 告警（可能是反压根子没治好，in-flight 数据持续堆积）；
• [ ] 确认 state backend 兼容性：RocksDB ✅、HashMap ❌（1.17）、Changelog ❌；
• [ ] 文档化 failover 恢复时长的变化，同步给 SRE。

一句话

checkpoint 对齐的代价，是「用快通道的反压换快照一致性」；unaligned checkpoint 的代价，是「用 state 存储换延迟」。两者是同一笔账的两种记法，选哪个取决于你的反压根子在哪、你的 state 还扛不扛得住。

注：本文的 Flink 版本为 1.17.2，参数与指标名以该版本为准；1.18/1.19 部分指标名有变化，请以官方文档为准。