Flink 增量 Checkpoint ，容错恢复后，随着时间的推移，之前的 Checkpoint
大家好，我现在有个任务，状态比较大，使用的增量的 Checkpoint，Flink 1.6 版本，默认的 Checkpoint 保留数为1，
我看了一下源码，发现当完成的 Checkpoint 数大于 state.checkpoints.num-retained的数值时，会对之前的完成的
Checkpoint 状态做清理。
当时我早上八点40多任务 Checkpoint 成功，当时的 chk-id 为
94040，然后到下午15点之间，Checkpoint都是超时失败的，然后15点容错恢复，从
chk-94040 恢复（八点40的状态），最新的 chk-id 为 94080。94080，94081，94082都是成功的，后面接着任务
Checkpoint 超时失败。
但此时 早上八点40多的任务的状态(也就是94040)还是没有清理，但是 state.checkpoints.num-retained又为1，完成的
Checkpoint 数大于1，所以我的理解，应该会清理掉 94040（早上八点40的状态），但是实际没有清理，状态文件还在 HDFS
上面。这是为什么呢，难道说状态容错恢复，不会清理之前的状态吗？
希望有大佬能帮我解惑，非常感谢*来自志愿者整理的flink邮件归档
写回答
-
如果你从chk-94040 进行checkpoint恢复的话，这个checkpoint是不会被删除清理的，这个行为是by design的。原因是因为从checkpoint resume在行为上被认为从Savepoint resume行为是一致的，也复用了一套代码 [1]，Savepoint的生命周期由用户把控，Flink框架自行不会去删除。
因此，加载的checkpoint被赋予了savepoint的property [2]。 这个CheckpointProperties#SAVEPOINT 里面的 discardSubsumed 属性是false，也就是当新的checkpoint完成时，在subsume阶段这个旧的checkpoint不会被discard掉，所以你restored的chk-94040是一直保留的。
希望这些解释能解答你的困惑
[1] https://github.com/apache/flink/blob/ee3101a075f681501fbc8c7cc4119476d497e5f3/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java#L1141
[2] https://github.com/apache/flink/blob/ee3101a075f681501fbc8c7cc4119476d497e5f3/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/Checkpoints.java#L214*来自志愿者整理的FLINK邮件归档
2021-12-02 18:01:57赞同 展开评论 打赏
版权声明：本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
