Flink 批作业 JobMaster Failover 进度恢复不再“JM 一挂，全盘重跑”

1. 背景：为什么批作业过去恢复不了进度？

在 JobMaster 故障被终止后，过去会出现两种情况：(nightlies.apache.org)

• HA 未开启：作业直接失败

• HA 已开启：会发生 JobMaster failover，作业重启

  • 流作业能从最近一次成功 checkpoint 恢复
  • 批作业通常没有 checkpoint，只能从头开始跑，之前进度全丢，这对长跑批作业是明显退步(nightlies.apache.org)

2. 核心原理：JobEventStore + 中间结果保留 + JM 重建状态

这套机制的关键在于两件事：

1. JobEventStore 记录“JobMaster 侧的状态变更事件”到外部文件系统
   例如 ExecutionGraph、OperatorCoordinator 等的状态变更会被记录下来。(nightlies.apache.org)

2. 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)

我建议你按下面方式理解它们（含默认值来自配置文档）。

一个直觉总结：

• 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)

1. 如果这个 Source 还有 task 没跑完：failover 后会重置并重跑该 Source 的所有 task
2. 如果该 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 min
execution.batch.job-recovery.previous-worker.recovery.timeout: 60 s
job-event.store.write-buffer.flush-interval: 1 s
job-event.store.write-buffer.size: 4 mb