Flink 1.16 Job Manager 重启后消息丢失问题排查及解决

Flink 作业在遇到异常时，会根据配置的重启策略进行自动重启。但如果整个 Job Manager 重启，可能会出现消息丢失的情况。本文旨在帮助你排查和解决 Flink 1.16 中 Job Manager 重启后消息丢失的问题，涵盖了可能的原因和相应的解决方案，确保数据处理的完整性。

问题分析

当 Flink 作业中的某个 Task 遇到异常，并触发配置的重启策略时，Flink 会尝试重启该 Task，恢复到最近一次 checkpoint 的状态，并重新处理 checkpoint 之后的数据。但如果整个 Job Manager 宕机并重启，则情况会更加复杂。以下是一些可能导致消息丢失的原因：

1. 死循环（Poison Pill）: 如果你的数据流中存在无法处理的“毒丸”（Poison Pill）数据，Flink 可能会陷入 fail -> restart -> fail again 的死循环。每次重启后，Flink 都会尝试处理该毒丸数据，导致作业持续失败。

2. Source 不支持 Checkpointing: 某些 Source 连接器可能不支持 Flink 的 checkpoint 机制。这意味着在 Job Manager 重启后，Source 无法回溯到上次 checkpoint 的位置，从而导致数据丢失。

3. Source 不可回溯: Flink 的容错机制依赖于 Source 的可回溯性。如果 Source 无法回溯，例如从 Socket 或 HTTP 端点读取数据，那么在重启后，将无法重新消费之前未完成的数据。

4. JobManagerCheckpointStorage: 如果你使用 JobManagerCheckpointStorage，那么 checkpoint 数据存储在 Job Manager 的内存中。当 Job Manager 重启后，这些 checkpoint 数据将会丢失，导致数据丢失。

5. 集群未配置高可用: 如果你的 Flink 集群没有配置高可用 (HA)，那么 Job Manager 宕机后，无法自动恢复，需要手动重启，并且状态数据可能会丢失。

解决方案

针对以上可能的原因，可以采取以下措施来解决 Flink Job Manager 重启后消息丢失的问题：

1. 处理 Poison Pill 数据:

   • 数据清洗: 在数据进入 Flink 作业之前，进行数据清洗，过滤掉可能导致异常的数据。
   • 错误处理: 在 Flink 作业中，使用 try-catch 块捕获异常，并对异常数据进行特殊处理，例如将其写入死信队列。
   • 跳过错误记录: 使用 side output 将错误记录输出到单独的流，并从主数据流中排除。

   以下代码示例展示了如何使用 try-catch 块处理异常数据：

   DataStream stream = ...;
   
   stream.map(record -> {
       try {
           // 处理 record
           return process(record);
       } catch (Exception e) {
           // 处理异常
           LOG.error("Error processing record: {}", record, e);
           // 可以选择返回一个默认值，或者抛出异常
           return null; // 如果返回 null，需要确保下游操作可以处理 null 值
       }
   }).filter(Objects::nonNull) // 过滤掉 null 值
   .sink(...);

2. 选择支持 Checkpointing 的 Source:

   

   ChatGPT Website Builder

   ChatGPT网站生成器，AI对话快速生成网站

   下载

   尽可能选择支持 Flink checkpoint 机制的 Source 连接器，例如 Apache Kafka Connector, Apache Pulsar Connector 等。这些连接器能够保证在重启后，能够从上次 checkpoint 的位置继续消费数据。

3. 确保 Source 可回溯:

   如果你的 Source 无法回溯，可以考虑使用 Flink 的 Exactly-Once 语义，结合事务性 Sink，例如 TwoPhaseCommitSinkFunction，来保证数据的完整性。

4. 使用持久化 Checkpoint 存储:

   不要使用 JobManagerCheckpointStorage，而是选择持久化的 checkpoint 存储，例如 HDFS, S3, RocksDB 等。这样即使 Job Manager 重启，checkpoint 数据也不会丢失。

   在 flink-conf.yaml 中配置 checkpoint 存储：

   state.backend: filesystem
   state.checkpoints.dir: hdfs:///flink/checkpoints
   state.savepoints.dir: hdfs:///flink/savepoints

5. 配置 Flink 集群高可用:

   配置 Flink 集群的高可用 (HA)，确保在 Job Manager 宕机后，能够自动切换到备用的 Job Manager，并从 checkpoint 恢复作业状态。

   关于 HA 的配置，请参考 Flink 官方文档：https://www.php.cn/link/3dd420c40e25463497c9fbaabf8b4621

注意事项

• 监控: 监控 Flink 作业的运行状态，及时发现并解决问题。
• 日志: 查看 Flink 的日志，了解作业的运行情况和错误信息。
• 测试: 在生产环境部署之前，进行充分的测试，确保作业的稳定性和可靠性。
• 版本: 使用稳定的 Flink 版本，并及时更新到最新的版本。

总结

解决 Flink Job Manager 重启后消息丢失的问题需要综合考虑多个方面，包括数据质量、Source 连接器的选择、checkpoint 存储的选择以及集群的高可用配置。通过合理的配置和优化，可以有效地避免消息丢失，确保 Flink 作业的稳定性和可靠性。