Flink 1.16 JobManager 重启后消息丢失问题排查与解决-java教程-PHP中文网

Flink 1.16 JobManager 重启后消息丢失问题排查与解决

霞舞

发布： 2025-10-08 10:48:39

原创

565人浏览过

flink 1.16 jobmanager 重启后消息丢失问题排查与解决

在 Flink 1.16 中，JobManager 重启后消息丢失是一个比较棘手的问题。以下将从多个角度分析可能的原因，并提供相应的解决方案。

首先，我们引用上面的摘要：本文针对 Flink 1.16 中遇到的 JobManager 重启后消息丢失问题，提供了一系列可能的排查方向和解决方案。文章涵盖了从检查是否陷入死循环、确认 Source 是否支持 Checkpointing 和 Rewindable，到排除 JobManagerCheckpointStorage 导致的 Checkpoint 丢失等多个方面，并提供了高可用配置的建议，旨在帮助读者全面理解并解决此类问题，确保 Flink 作业的稳定性和数据完整性。

1. 检查是否陷入 "Fail -> Restart -> Fail Again" 死循环

最常见的原因是程序遇到了“毒丸”（Poison Pill）数据，即无法被正确处理的记录。Flink 在遇到这种数据时，会不断地尝试重启并重新处理该数据，从而陷入死循环，导致后续消息无法被处理。

解决方案：

数据清洗： 在 Source 端进行数据清洗，过滤掉不符合格式或会导致异常的数据。
容错处理： 在算子中添加容错处理逻辑，例如使用 try-catch 捕获异常，并记录错误信息，避免程序崩溃。
侧输出流： 将无法处理的数据发送到侧输出流，进行后续分析和处理。

示例代码：

DataStream<String> input = ...;

DataStream<String> processed = input.map(value -> {
    try {
        // 尝试处理数据
        return process(value);
    } catch (Exception e) {
        // 记录错误信息
        LOG.error("Error processing value: {}", value, e);
        // 将数据发送到侧输出流
        return null; // 或者抛出异常，并使用侧输出流捕获
    }
}).filter(Objects::nonNull); // 过滤掉 null 值

// 获取侧输出流
OutputTag<String> errorTag = new OutputTag<String>("error-tag", Types.STRING);
DataStream<String> errorStream = processed.getSideOutput(errorTag);

登录后复制

2. Source 是否支持 Checkpointing 和 Rewindable

Flink 的容错机制依赖于 Checkpointing 和 Source 的 Rewindable 能力。如果 Source 不支持 Checkpointing，或者无法 Rewind 到上次 Checkpoint 的位置，则会导致数据丢失。

解决方案：

选择支持 Checkpointing 的 Source： 尽量选择 Flink 官方提供的或经过验证的、支持 Checkpointing 的 Source Connector。
自定义 Source： 如果必须使用不支持 Checkpointing 的 Source，可以考虑自定义 Source，并实现 Checkpointing 接口。需要注意的是，自定义 Source 的 Checkpointing 实现较为复杂，需要仔细考虑数据一致性和性能问题。
使用 Kafka 或其他可靠消息队列： 将数据先写入 Kafka 等可靠消息队列，再从 Kafka 读取数据进行处理。Kafka 具有持久化存储和回溯消费的能力，可以保证数据不丢失。

3. Checkpoint 存储位置

Checkpoint 的存储位置也会影响数据恢复。如果使用 JobManagerCheckpointStorage，则 Checkpoint 数据存储在 JobManager 的内存中。当 JobManager 重启时，Checkpoint 数据会丢失，导致数据无法恢复。

AI建筑知识问答

用人工智能ChatGPT帮你解答所有建筑问题

查看详情

解决方案：

配置高可用性存储： 使用高可用性的 Checkpoint 存储，例如 HDFS、RocksDB 或 S3。这些存储方式可以将 Checkpoint 数据持久化存储，即使 JobManager 重启，数据也不会丢失。

配置示例 (flink-conf.yaml):

state.checkpoints.dir: hdfs:///flink/checkpoints
state.savepoints.dir: hdfs:///flink/savepoints
state.backend: rocksdb
state.backend.rocksdb.memory.managed: true

登录后复制

4. JobManager 高可用配置

JobManager 的重启不应该导致数据丢失。如果 JobManager 重启导致数据丢失，说明集群的高可用配置可能存在问题。

解决方案：

配置 ZooKeeper 或其他高可用协调服务： 配置 ZooKeeper 或其他高可用协调服务，用于选举 Leader JobManager，并在 Leader JobManager 失败时自动切换到备用 JobManager。
配置高可用性存储： 如前所述，使用高可用性的 Checkpoint 存储，确保 Checkpoint 数据不丢失。

高可用配置示例 (flink-conf.yaml):

high-availability: zookeeper
high-availability.storageDir: hdfs:///flink/ha
high-availability.cluster-id: /flink-cluster
high-availability.zookeeper.quorum: zk-host1:2181,zk-host2:2181,zk-host3:2181

登录后复制

总结

解决 Flink JobManager 重启后消息丢失问题，需要从多个方面进行排查。首先要确定是否陷入死循环，然后检查 Source 是否支持 Checkpointing 和 Rewindable，接着排除 Checkpoint 存储位置的影响，最后配置 JobManager 的高可用性。通过以上步骤，可以有效地解决该问题，保证 Flink 作业的稳定性和数据完整性。

注意事项：