Flink 批作业 JobMaster Failover 进度恢复不再“JM 一挂，全盘重跑”
1. 背景：为什么批作业过去恢复不了进度？
在 JobMaster 故障被终止后，过去会出现两种情况：(nightlies.apache.org)
HA 未开启：作业直接失败
HA 已开启：会发生 JobMaster failover，作业重启
- 流作业能从最近一次成功 checkpoint 恢复
- 批作业通常没有 checkpoint，只能从头开始跑，之前进度全丢，这对长跑批作业是明显退步(nightlies.apache.org)
2. 核心原理：JobEventStore + 中间结果保留 + JM 重建状态
这套机制的关键在于两件事：
JobEventStore 记录“JobMaster 侧的状态变更事件”到外部文件系统
例如 ExecutionGraph、OperatorCoordinator 等的状态变更会被记录下来。(nightlies.apache.org)JM 挂掉到重启期间，TaskManager 继续保留已产出的中间结果并不断尝试重连
当新 JM 拉起后，会重新与 TM 建立连接，并基于：
- TM 保留的中间结果
- JobEventStore 中记录的事件
来恢复作业状态，从而继续推进批作业执行进度。(nightlies.apache.org)
你可以把它理解成：
- 以前：JM 一挂，批作业“失忆”，只能重头跑
- 现在：JM 一挂，TM 先把“半成品”捂住，外部 FS 里还有“操作日志”，新 JM 来了就能“续上”
3. 如何开启：两步就够（但前提很硬）
3.1 必须开启集群 HA
要启用该能力，首先必须开启 HA（Flink 支持 ZooKeeper 或 Kubernetes HA）。(nightlies.apache.org)
3.2 打开开关：execution.batch.job-recovery.enabled: true
配置项：execution.batch.job-recovery.enabled = true。(nightlies.apache.org)
补充两个关键事实：
- 该开关默认是
false（需要你显式开启）(nightlies.apache.org) - 目前只有 Adaptive Batch Scheduler 支持该能力(nightlies.apache.org)
而 batch 作业在未显式指定调度器时，默认会使用AdaptiveBatch（所以通常不用额外改）(nightlies.apache.org)
4. 调优参数：在“恢复更精细”与“I/O 压力”之间做权衡
官方给了 4 个核心调优点：(nightlies.apache.org)
我建议你按下面方式理解它们（含默认值来自配置文档）。
| 配置项 | 默认值 | 作用 | 调优方向（经验） |
|---|---|---|---|
execution.batch.job-recovery.snapshot.min-pause | 3 min | OperatorCoordinator / ShuffleMaster 做“快照”的最小间隔，用于避免过于频繁快照导致性能下降(nightlies.apache.org) | 想更小回退：调小（I/O 更高）；集群 I/O 紧张：调大 |
execution.batch.job-recovery.previous-worker.recovery.timeout | 30 s | 新 JM 等待“旧 worker（shuffle worker）重连”的超时；重连后会把已产出中间结果细节上报给新 JM，以便复用(nightlies.apache.org) | 大集群/网络抖动：适当调大；想更快进入恢复：调小但可能复用变少 |
job-event.store.write-buffer.flush-interval | 1 s | JobEventStore 写缓冲刷到外部 FS 的周期(nightlies.apache.org) | 外部 FS 延迟高：可适当调大；想更实时落盘：调小但 I/O 频繁 |
job-event.store.write-buffer.size | 1 mb | JobEventStore 写缓冲大小，满了就刷到外部 FS(nightlies.apache.org) | 事件量大/FS 吞吐好：可调大减少刷盘次数；太大可能导致“事件落盘延迟”更明显 |
一个直觉总结：
- snapshot.min-pause决定“恢复时最多回退多少进度（更准确地说：回退的粒度）”
- previous-worker.recovery.timeout决定“你愿意等 TM 把可复用的中间结果信息报全多久”
- JobEventStore 两个 buffer 参数决定“写外部 FS 的节奏与开销”
5. 让 Source 真正支持“进度恢复”：必须是 FLIP-27 新 Source
这点非常关键：该能力目前只对 FLIP-27 新 Source 生效。(nightlies.apache.org)
要让你的 Source 在 JM failover 后尽量不重跑，要求是：
- 新 Source 的SplitEnumerator必须支持在 batch 场景下做 state snapshot（此时
checkpointId = -1）并实现SupportsBatchSnapshot接口(nightlies.apache.org)
如果你的 Source 不支持，会发生什么？官方给了两条“保正确性”的兜底逻辑：(nightlies.apache.org)
- 如果这个 Source 还有 task 没跑完：failover 后会重置并重跑该 Source 的所有 task
- 如果该 Source 的所有 task 都已完成：作业可以继续
但如果未来某个时刻这些 task 需要重启（例如PartitionNotFound），那么还是会导致该 Source 的所有 subtasks 被重置并重跑(nightlies.apache.org)
实战建议（写 Source 的人很受用）：
snapshotState 里存什么：
- 未分配 splits
- 已分配但未完成的 splits（以及它们的进度/offset/position）
- 已完成 splits 的集合（用于恢复后跳过）
恢复时最重要的是：别重复发 splits，否则就是“看似恢复了，实则重复计算/重复读数据”
6. 局限性：别踩这 3 个坑
官方列的限制很明确：(nightlies.apache.org)
- 只支持新 Source（FLIP-27），legacy source 不支持(nightlies.apache.org)
- 仅适用于Adaptive Batch Scheduler（因此继承它的一切限制）(nightlies.apache.org)
- 不支持 remote shuffle services(nightlies.apache.org)
这点也和“TM 保留中间结果并上报给新 JM”这一恢复路径强相关(nightlies.apache.org)
7. 一份最小可用配置示例（思路版）
你可以按这个思路在flink-conf.yaml落地（HA 具体细节按你们是 ZK 还是 K8s HA 来配）：
# 1) 先确保 HA 已启用（ZK 或 K8s HA）# 2) 开启批作业 JM failover 进度恢复execution.batch.job-recovery.enabled:true# 3) 可选：按需调优execution.batch.job-recovery.snapshot.min-pause:1 minexecution.batch.job-recovery.previous-worker.recovery.timeout:60 sjob-event.store.write-buffer.flush-interval:1 sjob-event.store.write-buffer.size:4 mb