Flink 作业在遇到异常时，会根据配置的重启策略进行自动重启。但如果整个 Job Manager 重启，可能会出现消息丢失的情况。本文旨在帮助你排查和解决 Flink 1.16 中 Job Manager 重启后消息丢失的问题，涵盖了可能的原因和相应的解决方案，确保数据处理的完整性。
问题分析
当 Flink 作业中的某个 Task 遇到异常，并触发配置的重启策略时，Flink 会尝试重启该 Task，恢复到最近一次 checkpoint 的状态，并重新处理 checkpoint 之后的数据。但如果整个 Job Manager 宕机并重启，则情况会更加复杂。以下是一些可能导致消息丢失的原因：
死循环（Poison Pill）: 如果你的数据流中存在无法处理的“毒丸”（Poison Pill）数据，Flink 可能会陷入 fail -> restart -> fail again 的死循环。每次重启后，Flink 都会尝试处理该毒丸数据，导致作业持续失败。
Source 不支持 Checkpointing: 某些 Source 连接器可能不支持 Flink 的 checkpoint 机制。这意味着在 Job Manager 重启后，Source 无法回溯到上次 checkpoint 的位置，从而导致数据丢失。
Source 不可回溯: Flink 的容错机制依赖于 Source 的可回溯性。如果 Source 无法回溯，例如从 Socket 或 HTTP 端点读取数据，那么在重启后，将无法重新消费之前未完成的数据。
JobManagerCheckpointStorage: 如果你使用 JobManagerCheckpointStorage，那么 checkpoint 数据存储在 Job Manager 的内存中。当 Job Manager 重启后，这些 checkpoint 数据将会丢失，导致数据丢失。
集群未配置高可用: 如果你的 Flink 集群没有配置高可用 (HA)，那么 Job Manager 宕机后，无法自动恢复，需要手动重启，并且状态数据可能会丢失。
解决方案
针对以上可能的原因，可以采取以下措施来解决 Flink Job Manager 重启后消息丢失的问题：
-
处理 Poison Pill 数据:
- 数据清洗: 在数据进入 Flink 作业之前，进行数据清洗，过滤掉可能导致异常的数据。
- 错误处理: 在 Flink 作业中，使用 try-catch 块捕获异常，并对异常数据进行特殊处理，例如将其写入死信队列。
- 跳过错误记录: 使用 side output 将错误记录输出到单独的流，并从主数据流中排除。
以下代码示例展示了如何使用 try-catch 块处理异常数据：
DataStream<YourDataType> 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(...); -
选择支持 Checkpointing 的 Source:
尽可能选择支持 Flink checkpoint 机制的 Source 连接器，例如 Apache Kafka Connector, Apache Pulsar Connector 等。这些连接器能够保证在重启后，能够从上次 checkpoint 的位置继续消费数据。
-
确保 Source 可回溯:
如果你的 Source 无法回溯，可以考虑使用 Flink 的 Exactly-Once 语义，结合事务性 Sink，例如 TwoPhaseCommitSinkFunction，来保证数据的完整性。
-
使用持久化 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
-
配置 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 作业的稳定性和可靠性。
