Flink 1.13.3自定义文件源Checkpoint异常及任务状态验证咨询

我基于原生Flink File Source实现了自定义文件源，通过自定义策略调用signalsNoSplits自动终止Flink应用。受外部限制，必须使用Flink 1.13.3的流处理模式（无界输入），无法使用批处理模式或相关流处理模式（因存在已知BUG）。

自定义文件源策略：监听Checkpoint Start和Complete调用，等待FileSplitAssigner.remainingSplits().size()为0（即最后一个文件分片已分配）后，额外等待N次（如4次）Checkpoint完成通知（遵循Checkpoint Subsuming协议），之后向源读取器发送signalsNoMoreSplits。

应用其他细节：未启用Unaligned checkpoints，拓扑为Custom File Source(S3) -> Process Function -> Native File Sink(S3)。

遇到的问题：最后一个文件分片发送后启动的Checkpoint在下游算子仍在处理记录时就已完成。根据Checkpoint文档，Checkpoint barrier事件应明确划分checkpoint前后的读取事件，预期此类Checkpoint需等所有算子处理完记录后才完成，目前因该问题导致输出存在丢失记录。

补充说明：自定义文件源还实现了“Ordered Source”，即按顺序读取源数据（如先读取种子数据初始化作业状态，再处理常规数据），采用上述相同策略。

咨询问题 #

1. 我对Checkpoint机制的理解是否存在偏差？
2. 还有哪些方式可以验证作业是否已完成所有记录读取且源或其他算子已处于非活跃状态？

1. Checkpoint机制理解的偏差 #

你的理解存在关键偏差：

• Checkpoint的核心目标是完成所有算子的状态快照，而非等待所有记录处理完毕。Checkpoint barrier的作用是划分快照的边界，确保快照仅包含barrier到达前处理的数据对应的状态，而不是触发下游算子处理完所有在途记录。
• 你混淆了“分片分配完毕”和“所有记录发送完毕”：FileSplitAssigner.remainingSplits().size() == 0仅表示没有新的分片需要分配，但当前正在处理的分片可能还有大量记录未读取、未发送到下游。此时触发的Checkpoint，Source会在发送完当前内存中的记录后插入barrier，随后继续读取剩余记录——这意味着Checkpoint完成时，下游算子可能还在处理该分片的后续记录，自然会导致数据丢失。
• 未启用Unaligned Checkpoint时，Checkpoint的完成条件是所有算子完成自身的状态快照，而非所有记录处理完成。哪怕下游还有记录在处理，只要算子完成了barrier之前的状态快照，Checkpoint就会被标记为完成。

2. 验证作业完成状态的可行方式 #

• 监控Flink UI指标：
  • 观察Source算子的numRecordsOut，该指标停止增长时，说明Source已发送完所有记录。
  • 查看Process Function和Sink的numRecordsIn/numRecordsOut，当这些指标不再变化，且Sink的pendingRecords（若有）为0时，代表下游已处理完所有数据。
• 自定义状态追踪：
  • 在Source中维护状态，不仅记录分片是否分配完毕，还要确认最后一个分片的所有记录已读取并发送完成。
  • 在Process和Sink算子中添加状态标记，记录是否还有未处理的在途记录。通过Checkpoint持久化这些状态，当所有状态都标记为“完成”时，即可确认作业处理完毕。
• 算子空闲状态检查：
  • 通过Flink REST API查询每个算子的isIdle状态，当所有算子都处于Idle状态，且作业仍为RUNNING但无数据流动时，说明已无活跃任务。
• Sink端验证：
  • 针对S3 Sink，检查是否存在.inprogress后缀的临时文件，且已完成的文件大小与预期一致——无临时文件且文件完整，说明Sink已完成所有写入。

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