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

本文旨在帮助你分析可能在使用 Flink 1.16 时，配置了重启策略后，JobManager 在达到最大重试次数后重启，导致部分消息丢失的问题的原因，并提供相应的解决方案，确保 Flink 应用在发生故障时能够可靠地处理数据，保障数据处理的完整性。

可能的原因及解决方案

在排查 Flink JobManager 重启导致消息丢失的问题时，需要从多个方面进行分析，以下是一些常见的原因及相应的解决方案：

1. 陷入 fail -> restart -> fail again 循环

问题描述： 如果你的 Flink 应用遇到无法处理的“毒丸”（poison pill）数据，会导致任务不断失败、重启，但始终无法跳过该数据。

解决方案：

• 数据清洗： 在数据进入 Flink 之前，进行数据清洗，过滤掉不符合规范或可能导致异常的数据。
• 异常处理： 在 Flink 应用中添加异常处理逻辑，捕获特定类型的异常，并采取相应的措施，例如跳过错误数据或将其发送到死信队列。
• flink.checkpoint.ignore-unrecoverable-state配置: 在flink-conf.yaml配置文件中将此参数设置为true, 可以跳过无法恢复的状态。

示例代码：

DataStream<YourDataType> stream = env.addSource(new YourSourceFunction())
    .map(data -> {
        try {
            // 数据处理逻辑
            return processData(data);
        } catch (Exception e) {
            // 异常处理逻辑
            LOG.error("Error processing data: {}", data, e);
            // 可以选择跳过当前数据，或者将其发送到死信队列
            return null; // 如果返回 null，需要确保后续算子能够处理 null 值
        }
    })
    .filter(Objects::nonNull); // 过滤掉 null 值

注意事项： 在选择跳过错误数据时，需要仔细评估其对业务的影响，确保不会造成数据不一致或其他问题。

2. Source 不支持 Checkpointing

问题描述： 如果你使用的 Source Function 没有实现 Checkpointing 接口，或者没有正确地维护状态，那么在 JobManager 重启后，可能会丢失部分数据。

解决方案：

• 使用支持 Checkpointing 的 Source： 尽可能使用 Flink 官方提供的或者经过验证的支持 Checkpointing 的 Source Function。
• 自定义 Source Function： 如果需要使用自定义的 Source Function，请确保其实现了 CheckpointedFunction 或 SourceFunction.SourceContext 接口，并正确地维护状态。

示例代码（自定义 Source Function）：

public class CustomSourceFunction implements SourceFunction<YourDataType>, CheckpointedFunction {

    private ListState<Long> offsetState;
    private long offset = 0;
    private volatile boolean isRunning = true;

    @Override
    public void run(SourceContext<YourDataType> ctx) throws Exception {
        while (isRunning) {
            // 从数据源读取数据
            YourDataType data = fetchData(offset);

            // 将数据发送到下游
            ctx.collect(data);

            // 更新 offset
            offset++;

            // 暂停一段时间
            Thread.sleep(100);
        }
    }

    @Override
    public void cancel() {
        isRunning = false;
    }

    @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
        offsetState.clear();
        offsetState.add(offset);
    }

    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        ListStateDescriptor<Long> descriptor =
                new ListStateDescriptor<>(
                        "offset-state",
                        TypeInformation.of(Long.class));

        offsetState = context.getOperatorStateStore().getListState(descriptor);

        if (context.isRestored()) {
            for (Long offset : offsetState.get()) {
                this.offset = offset;
            }
        }
    }

    private YourDataType fetchData(long offset) {
        // 从数据源读取数据的逻辑
        // ...
        return null;
    }
}

注意事项： 在实现 CheckpointedFunction 接口时，需要注意状态的序列化和反序列化，以及状态的备份和恢复。

3. Source 不支持 Rewind

问题描述： Flink 的容错机制依赖于 Source Function 能够回溯到上一个 Checkpoint 的位置，重新消费数据。如果 Source Function 不支持回溯（例如，从 Socket 或 HTTP 端点读取数据），那么在 JobManager 重启后，可能会丢失部分数据。

AI建筑知识问答

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

22

查看详情

解决方案：

• 使用支持 Rewind 的 Source： 尽可能使用支持回溯的 Source Function，例如 Kafka Connector。
• 自定义 Source Function： 如果需要使用自定义的 Source Function，可以考虑使用类似于 Kafka 的消息队列作为中间层，实现数据的持久化和回溯。

4. 使用 JobManagerCheckpointStorage

问题描述： 如果你使用了 JobManagerCheckpointStorage，那么 Checkpoint 数据会存储在 JobManager 的内存中。当 JobManager 重启后，Checkpoint 数据会丢失，导致 Flink 应用无法恢复到之前的状态。

解决方案：

• 使用持久化的 Checkpoint Storage： 建议使用持久化的 Checkpoint Storage，例如 FileSystemCheckpointStorage 或 RocksDBCheckpointStorage，将 Checkpoint 数据存储在 HDFS 或 RocksDB 中，确保在 JobManager 重启后数据不会丢失。

配置示例：

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

注意事项： 使用持久化的 Checkpoint Storage 会增加 Flink 应用的 I/O 开销，需要根据实际情况进行权衡。

5. JobManager 频繁重启

问题描述： JobManager 的频繁重启本身就是一个需要解决的问题。JobManager 的职责是管理 Flink 集群的资源和任务，如果 JobManager 频繁重启，会导致 Flink 应用不稳定，甚至无法正常运行。

解决方案：

• 排查 JobManager 的日志： 查看 JobManager 的日志，分析导致其重启的原因。
• 调整 JVM 参数： 适当调整 JobManager 的 JVM 参数，例如堆大小和 GC 策略，避免内存溢出或频繁 GC。
• 升级 Flink 版本： 升级到最新的 Flink 版本，可以修复一些已知的 Bug 和性能问题。
• 配置高可用性： 配置 Flink 集群的高可用性，确保在 JobManager 发生故障时，能够自动切换到备用的 JobManager，避免单点故障。 具体可以参考官方文档配置高可用性

总结：

解决 Flink JobManager 重启导致消息丢失的问题需要综合考虑多个方面，包括数据清洗、异常处理、Source Function 的选择和实现、Checkpoint Storage 的配置以及 JobManager 的稳定性。通过仔细分析问题的原因，并采取相应的解决方案，可以确保 Flink 应用在发生故障时能够可靠地处理数据，保障数据处理的完整性。

以上就是Flink 1.16 JobManager 重启导致消息丢失问题排查与解决的详细内容，更多请关注php中文网其它相关文章！