📚 课程目录第4章端到端 Exactly-Once 实战

4、端到端 Exactly-Once 实战：Kafka Source + Flink + Kafka Sink 的完整链路配置与验证

好，前面我们把理论、两阶段提交、状态后端都聊透了。这一章，咱们直接上硬菜——把 Kafka Source、Flink 作业、Kafka Sink 串起来，跑一个真正的端到端 Exactly-Once 链路。

说实话，很多同学在面试时能把原理讲得头头是道，但一到实际配置就翻车。我当年第一次搭这个链路时，就因为漏了一个 transactional.id 前缀配置，数据重复了整整一天。嗯，从那以后我养成了一个习惯：每次上线前，先跑一个小流量验证链路。

4.1 链路全景：三个组件如何协同

先看整体架构。这条链路涉及三个角色：

Kafka Source：负责从源头 Topic 消费数据，并定期保存 Offset 到 Flink 的 Checkpoint 中。
Flink 作业：处理逻辑（过滤、转换、聚合），同时管理 Checkpoint 和两阶段提交的协调。
Kafka Sink：利用 Flink 的 ExactlyOnceSinkFunction 或 KafkaProducer 的事务 API，将数据写入目标 Topic。

它们之间的协作，说白了就是靠 Checkpoint 这根「指挥棒」。每次 Checkpoint 触发时，Source 保存 Offset，Sink 预提交事务，JobManager 确认所有算子都准备好了，才正式提交事务。任何一个环节失败，整个 Checkpoint 就回滚，数据不会丢也不会重复。

核心原则：端到端 Exactly-Once 不是某个组件的功劳，而是 Source、Flink、Sink 三者通过 Checkpoint 机制达成的「分布式共识」。

下面这张图展示了完整的流程：

4.2 环境准备与版本匹配

动手之前，先确认版本。我踩过最大的坑就是版本不兼容。Flink 1.12 之前和之后的 Kafka 连接器 API 变化很大，你要是用 Flink 1.14 配一个旧版 Kafka 客户端，运行时直接报 NoSuchMethodError。

组件	推荐版本	说明
Flink	1.14+	内置 Kafka Connector 2.0+，支持新 API
Kafka	2.8+	需要开启事务功能（默认开启）
Kafka Client	2.8+	与 Broker 版本保持一致
ZooKeeper / KRaft	3.5+ / 3.0+	KRaft 模式无需 ZK，但 Flink 连接器兼容

小提示：我个人习惯在 Maven 或 Gradle 中显式指定 Kafka Client 版本，避免传递依赖带来不兼容的版本。比如 Flink 1.14 默认带的 Kafka Client 是 2.8.0，你如果用了 3.0 的 Broker，最好手动升级 Client 到 3.0。

4.3 Source 端配置：开启 Offset 精确保存

Source 端要做到 Exactly-Once，核心是让 Flink 自己管理 Offset，而不是交给 Kafka 的消费者组。你想想看，如果让 Kafka 自动提交 Offset，那 Flink 恢复时读到的 Offset 可能已经提交了，但数据还没处理完——这就丢数据了。

正确的做法是：关闭 Kafka 消费者的自动提交，让 Flink 把 Offset 存到 Checkpoint 里。

// Java 示例：KafkaSource 配置
KafkaSource<String> source = KafkaSource.<String>builder()
    .setBootstrapServers("localhost:9092")
    .setTopics("input-topic")
    .setGroupId("flink-exactly-once-group")
    .setStartingOffsets(OffsetsInitializer.earliest())
    // 关键：关闭自动提交
    .setProperty("enable.auto.commit", "false")
    // 设置事务超时时间，必须小于 Kafka 的 max.transaction.timeout.ms
    .setProperty("transaction.timeout.ms", "60000")
    .setValueOnlyDeserializer(new SimpleStringSchema())
    .build();

这里有个细节：transaction.timeout.ms 默认是 15 分钟，但 Kafka Broker 的 max.transaction.timeout.ms 默认也是 15 分钟。如果你在 Flink 里设置超过 15 分钟，Kafka 会直接拒绝。我建议设成 60 秒，够用了。

4.4 Sink 端配置：事务写入与两阶段提交

Sink 端是 Exactly-Once 最难搞的地方。Flink 提供了 KafkaSink（新 API）和 FlinkKafkaProducer（旧 API），我强烈建议用新的 KafkaSink，它原生支持 Exactly-Once 语义。

// Java 示例：KafkaSink 配置
KafkaSink<String> sink = KafkaSink.<String>builder()
    .setBootstrapServers("localhost:9092")
    .setRecordSerializer(KafkaRecordSerializationSchema.builder()
        .setTopic("output-topic")
        .setValueSerializationSchema(new SimpleStringSchema())
        .build()
    )
    // 开启 Exactly-Once 语义
    .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
    // 设置事务前缀，必须全局唯一
    .setTransactionalIdPrefix("flink-sink-")
    .setProperty("transaction.timeout.ms", "60000")
    .build();

注意 setTransactionalIdPrefix，这个前缀用来生成 Kafka 事务 ID。如果你有多个 Flink 作业写入同一个 Topic，前缀必须不同，否则事务 ID 冲突，Kafka 会报 ProducerFencedException。我曾经在生产环境遇到过这个问题，两个作业用了同样的前缀，结果一个作业把另一个作业的事务给「 fencing 」掉了，数据直接写不进去。

4.5 Checkpoint 配置：调优参数

Checkpoint 是整个链路的「心脏」。配置不对，Exactly-Once 就是空谈。下面是我常用的配置：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 开启 Checkpoint
env.enableCheckpointing(10000); // 10 秒一次

// 设置 Exactly-Once 模式
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

// 超时时间
env.getCheckpointConfig().setCheckpointTimeout(60000);

// 同时进行的 Checkpoint 数量
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

// 两次 Checkpoint 之间的最小间隔
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(5000);

// 外部化 Checkpoint（作业取消后保留）
env.getCheckpointConfig().enableExternalizedCheckpoints(
    ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
);

// 设置状态后端（推荐 RocksDB）
env.setStateBackend(new HashMapStateBackend());
env.getCheckpointConfig().setCheckpointStorage("file:///flink/checkpoints");

几个关键点：

Checkpoint 间隔：10 秒是经验值。太短（比如 1 秒）会导致频繁的事务开启和提交，Kafka 压力大；太长（比如 5 分钟）会导致数据延迟高。
最小间隔：设成 5 秒，防止 Checkpoint 连续触发，给系统一点喘息时间。
外部化 Checkpoint：这个一定要开。作业意外取消时，Checkpoint 不会被删除，你可以从上次的位置恢复。

警告：如果你用 RocksDB 作为状态后端，记得检查磁盘空间。RocksDB 在 Checkpoint 时会生成大量临时文件，我曾经有一次因为磁盘写满导致 Checkpoint 连续失败，作业最终挂了。建议把 Checkpoint 目录放在独立的磁盘分区上。

4.6 完整作业代码示例

把上面几块拼起来，就是一个完整的端到端 Exactly-Once 作业：

public class ExactlyOnceKafkaJob {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // 1. Checkpoint 配置
        env.enableCheckpointing(10000);
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(5000);
        env.getCheckpointConfig().enableExternalizedCheckpoints(
            ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
        );
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("file:///flink/checkpoints");
        
        // 2. Kafka Source
        KafkaSource<String> source = KafkaSource.<String>builder()
            .setBootstrapServers("localhost:9092")
            .setTopics("input-topic")
            .setGroupId("flink-exactly-once-group")
            .setStartingOffsets(OffsetsInitializer.earliest())
            .setProperty("enable.auto.commit", "false")
            .setProperty("transaction.timeout.ms", "60000")
            .setValueOnlyDeserializer(new SimpleStringSchema())
            .build();
        
        // 3. 处理逻辑（示例：简单转换）
        DataStream<String> processed = env
            .fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source")
            .map(value -> "processed: " + value);
        
        // 4. Kafka Sink
        KafkaSink<String> sink = KafkaSink.<String>builder()
            .setBootstrapServers("localhost:9092")
            .setRecordSerializer(KafkaRecordSerializationSchema.builder()
                .setTopic("output-topic")
                .setValueSerializationSchema(new SimpleStringSchema())
                .build()
            )
            .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
            .setTransactionalIdPrefix("flink-sink-")
            .setProperty("transaction.timeout.ms", "60000")
            .build();
        
        processed.sinkTo(sink);
        
        env.execute("Exactly-Once Kafka Job");
    }
}

4.7 验证方法：如何确认 Exactly-Once 生效

配置写完了，怎么验证它真的做到了 Exactly-Once？我一般用三步验证法：

数据计数验证：往 Source Topic 发送 N 条数据，等作业处理完后，检查 Sink Topic 的数据量。如果也是 N 条，说明没有丢数据。但这个方法不能发现重复数据。
幂等性验证：在 Sink Topic 里加一个去重逻辑（比如用 Redis 记录已处理的 Key），然后故意触发一次作业故障恢复。恢复后，检查是否有重复的 Key 写入。如果没有，说明 Exactly-Once 生效。
事务日志验证：查看 Kafka 的 __transaction_state 内部 Topic，确认事务状态是 CompleteCommit 而不是 CompleteAbort。如果看到大量 Abort，说明事务提交失败了。

我的经验：最靠谱的验证方法是「故障注入」。在作业运行中手动杀掉 TaskManager，然后观察恢复后的数据。我一般会写一个脚本，每 30 秒随机 kill 一个 TaskManager，同时往 Source 里灌数据。跑 10 分钟后，检查 Sink 的数据是否完整且无重复。这个方法虽然粗暴，但能暴露 90% 的配置问题。

4.8 常见问题与避坑指南

最后，分享几个我踩过的坑：

事务超时导致作业失败：Checkpoint 间隔 + 处理时间必须小于 transaction.timeout.ms。如果 Checkpoint 需要 30 秒，但事务超时设了 20 秒，那每次 Checkpoint 都会失败。解决办法：增大事务超时，或者缩短 Checkpoint 间隔。
Kafka 事务 ID 冲突：多个作业或并行子任务使用相同的事务 ID 前缀，导致 ProducerFencedException。解决办法：每个作业用不同的前缀，或者在前缀里加上作业名称。
Checkpoint 存储空间不足：RocksDB 的 Checkpoint 文件会占用大量磁盘。我建议定期清理旧的 Checkpoint，或者使用 HDFS/S3 作为远程存储。
Kafka 版本不兼容：Flink 1.14 的 Kafka Connector 默认用 2.8.0 客户端，如果你用 Kafka 3.0+，需要手动升级客户端版本，否则会报 UnsupportedVersionException。

嗯，这一章的内容就到这。配置和验证方法都给了，剩下的就是动手实践。记住，理论再完美，不跑一遍永远不知道坑在哪。