Flink（S3后端）因state.checkpoints.num-retained引发Checkpoint触发延迟

环境与核心配置 #

我们在EMR集群运行Flink 1.17.0任务，以S3作为Checkpoint后端，任务总并行度5.5K，关键配置如下：

checkpoint-interval : 70000
min-pause-between-checkpoint : 15000
max-concurrent-checkpoint : 1
checkpoint-type: unaligned
TolerableCheckpointFailureNumber :2
Checkpoint-mode : atleast-once
incremental-checkpointing: true
state.backend: filesystem
state.checkpoints.dir: s3p://checkpoint-sp-test/_entropy_/
s3.entropy.key: _entropy_
s3.entropy.length: 1
presto.s3.connect-timeout: 1m
presto.s3.socket-timeout: 1m
presto.s3.max-connections: 6500
state.checkpoints.num-retained: 3
high-availability.type: zookeeper

问题现象与定位 #

按配置理论上Checkpoint应每70秒触发一次，但实际每次触发存在5-20分钟的延迟。

已确认延迟原因：每次触发新Checkpoint前，JobManager会执行阻塞式的旧Checkpoint删除流程，该逻辑由state.checkpoints.num-retained=3控制。将该值设为极大数（如6000000）时，Checkpoint能按正常间隔触发，但会引发新问题：新JobManager选举时会尝试加载配置指定数量的旧Checkpoint，因数量过大导致任务无法启动。

曾尝试通过S3生命周期规则自动清理1天前的Checkpoint，但需配合合理的num-retained配置解决矛盾。

可行解决方案 #

1. 适中num-retained值 + S3生命周期规则组合 #

• 选择一个平衡删除压力与JobManager加载量的数值（比如保留20-50个Checkpoint）：按70秒/次的生成频率，1天约产生1200个Checkpoint，保留20个的话，每次仅需删除1个过期Checkpoint，耗时可忽略。
• 配置S3生命周期规则，自动清理早于1天的Checkpoint，避免存储无限膨胀。
• 优势：无需修改代码，仅调整配置和S3规则，风险低；JobManager选举时加载的Checkpoint数量可控，不会启动失败。

2. 启用Checkpoint异步删除（Flink 1.16+支持） #

Flink 1.16及以上版本支持异步删除过期Checkpoint，避免阻塞新Checkpoint触发流程，添加以下配置：

state.checkpoints.async-delete: true

• 原理：将旧Checkpoint的删除操作放入异步线程执行，不占用Checkpoint触发的主线程资源。
• 注意：仍需将state.checkpoints.num-retained设为合理值，配合S3生命周期规则控制存储膨胀。

3. 优化S3客户端删除性能 #

针对S3删除慢的问题，调整相关配置提升效率：

• 适当调高presto.s3.max-connections（当前已设为6500，可根据EMR集群资源上限微调），增加并行删除的连接数。
• 确认Flink 1.17已启用S3批量删除API：FileSystem后端默认支持批量删除，可大幅减少API调用次数，提升删除速度。
• 调整s3.entropy.length：当前值为1会生成多个前缀，删除时需遍历多路径文件。若业务允许，可增大熵值长度减少前缀数量，或取消熵值配置（需注意S3单前缀的性能限制）。

4. 自定义Checkpoint清理逻辑（进阶） #

若上述方案无法满足需求，可通过Flink扩展机制实现自定义清理：

• 实现CheckpointListener，在Checkpoint成功完成后异步删除过期的Checkpoint，而非在触发前同步执行。
• 自定义元数据管理逻辑，仅保留最新N个Checkpoint的元数据，JobManager选举时仅加载这些记录，避免加载过多历史数据。

内容的提问来源于stack exchange，提问作者priyadhingra19