Flink 1.16任务重启后Task Managers未重处理故障数据求助

以下是针对你遇到的问题的排查方向和解决方案：

1. 缺失持久化Checkpoint存储配置 #

你的代码中注释掉了setCheckpointStorage配置项，这是核心问题之一：

// streamExecutionEnvironment.getCheckpointConfig().setCheckpointStorage("file:///opt/flink/checkpoint-dir/");

• 问题原因：未配置持久化的Checkpoint存储时，Flink默认将Checkpoint存储在JobManager的堆内存中。当作业重启时，这部分内存中的Checkpoint会丢失，作业无法从故障前的Checkpoint恢复，而是从数据源的最新提交偏移量开始消费，跳过了未处理完成的payload1。
• 解决方案：取消注释并配置一个持久化的存储路径（本地文件系统或分布式文件系统如HDFS），确保路径存在且Flink有读写权限：

  streamExecutionEnvironment.getCheckpointConfig().setCheckpointStorage("file:///opt/flink/checkpoint-dir/");
  

2. 数据源未正确支持Checkpoint或提前提交偏移量 #

你使用的sourceFactory.readSignalStream数据源需要兼容Flink的Checkpoint机制：

• 问题原因：如果数据源在读取数据后立即提交偏移量到外部系统（比如Kafka的自动提交），而不是等待Checkpoint完成后再提交，那么重启后数据源会从已提交的偏移量开始消费，不会重处理payload1。
• 解决方案：确保数据源使用Flink的Checkpoint来管理偏移量。例如对于Kafka数据源，需关闭自动提交，由Flink控制偏移量提交：

  properties.setProperty("enable.auto.commit", "false");
  

3. 故障发生前无成功完成的Checkpoint #

如果作业在第一个Checkpoint完成前就发生故障，那么没有可用的Checkpoint来恢复状态：

• 问题原因：Checkpoint需要时间完成，若故障发生在Checkpoint完成前，最新的Checkpoint会被丢弃，作业只能从头开始消费。
• 解决方案：
  • 查看Flink日志，搜索Completed checkpoint关键字，确认是否有成功完成的Checkpoint。
  • 调整Checkpoint频率（当前为5秒）或超时时间（当前为10分钟），确保Checkpoint能正常完成。

4. 验证故障发生时机与Checkpoint的关系 #

如果payload1已经处理完成且Checkpoint已成功提交后才发生异常，那么重启后不会重处理该数据：

• 问题原因：Checkpoint会记录当前的状态和数据源偏移量，若故障发生在Checkpoint完成之后，恢复时会从Checkpoint的位置继续，不会重复处理已完成的数据。
• 解决方案：对比日志中Checkpoint完成时间和异常发生时间，确认故障是否发生在处理payload1的过程中而非之后。

调试步骤 #

• 查看Flink Web UI的Checkpoint页面，确认Latest Completed Checkpoint存在且状态正常。
• 检查TaskManager和JobManager日志，定位异常发生的时间点与Checkpoint完成时间的先后关系。
• 验证数据源的偏移量管理逻辑，确认偏移量是否仅在Checkpoint完成后才提交。

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