Flink CDC mongo cdc checkpoint 存储的 state 有清理机制吗？

Flink CDC MongoDB CDC 的 checkpoint 存储的 state 确实存在清理机制，但其具体行为和配置需要结合 Flink 的状态管理和 MongoDB CDC 的特性来理解。以下是详细解答：

1. Checkpoint 中 State 的存储与清理机制

在 Flink 中，checkpoint 是用于保存应用程序状态（state）的快照机制，而 MongoDB CDC 的状态信息（如 Change Stream 的 Resume Token）会被存储在 checkpoint 中。Flink 提供了以下几种清理机制：

(1) TTL（Time-to-Live）机制

• 如果 MongoDB CDC 的状态算子（如 ChangelogNormalize）支持生命周期管理（TTL），则可以通过设置 table.exec.state.ttl 参数来定义状态的存活时间。
• 超过 TTL 时间的状态会被自动清理，从而避免状态无限增长导致资源耗尽。

(2) 手动清理

• 手动创建的 checkpoint 快照不会被自动清理规则覆盖，必须通过手动删除操作来清理。
• 如果作业长期停止，系统会在启动时根据之前配置的清理规则对历史快照进行清理。

(3) 自动清理规则

• 可以通过配置 历史快照保存个数 或 历史快照保存最长时间 来定义自动清理策略。
• 满足任一条件（如超过保存个数或保存时间）后，系统会自动清理旧的 checkpoint 快照。

2. MongoDB CDC 特有的 Resume Token 清理机制

MongoDB CDC 使用 Change Stream 的 Resume Token 来记录数据变更的位置，Resume Token 对应于 MongoDB 的 oplog.rs 集合中的位置。以下是其清理机制的关键点：

(1) oplog.rs 的保留时间

• Resume Token 的有效性依赖于 oplog.rs 集合中对应的记录是否存在。
• 如果 oplog.rs 的保留时间过短，可能会导致 Resume Token 失效，出现无效 Resume Token 的异常。
• 建议：通过调整 oplog.rs 的集合大小来延长其保留时间，确保 Resume Token 在 checkpoint 中的有效性。

(2) 新变更记录刷新 Resume Token

• 新到达的变更记录和心跳记录会不断刷新 Resume Token，因此即使旧的 Resume Token 被清理，新的 Token 仍然可以用于恢复。

3. 企业级状态后端的优化

在 VVR 8.0.1 引擎版本中，Flink 的企业级状态后端进行了重构，进一步提升了状态清理的效率： - 更紧凑的状态格式：减少了本地状态存储空间，降低了存储开销。 - 高效的数据清理策略：显著提升了状态访问性能，并优化了大状态场景下的扩缩容和恢复机制。

4. 重要提醒

• 手动创建的 checkpoint 快照无法通过自动清理规则删除，需要定期手动清理，否则可能导致不必要的存储费用。
• oplog.rs 的大小配置需谨慎，避免因保留时间过短导致 Resume Token 失效。
• 建议启用 TTL 机制，以自动清理过期状态，减少资源占用。

综上所述，Flink CDC MongoDB CDC 的 checkpoint 存储的 state 具备多种清理机制，包括 TTL、手动清理、自动清理规则以及 oplog.rs 的保留时间管理。合理配置这些机制可以有效避免状态无限增长带来的性能问题和资源浪费。