Flink Kafka 端到端 Exactly - HelloWorld开发者社区

摘要： 本文基于 Flink 1.9.0 和 Kafka 2.3 版本，对 Flink kafka 端到端 Exactly-Once 进行分析及 notifyCheckpointComplete 顺序，主要内容分为以下两部分：

1.Flink-kafka 两阶段提交源码分析

TwoPhaseCommitSinkFunction 分析

2.Flink 中 notifyCheckpointComplete 方法调用顺序

定义
样例
operator 调用 notifyCheckpointComplete
对 Exactly-Once 语义的影响

Tips：Flink 中文社区征稿啦，感兴趣的同学可点击「阅读原文」了解详情～

Flink-kafka 两阶段提交源码分析

FlinkKafkaProducer 实现了 TwoPhaseCommitSinkFunction，也就是两阶段提交。关于两阶段提交的原理，可以参见《An Overview of End-to-End Exactly-Once Processing in Apache Flink》，本文不再赘述两阶段提交的原理，但是会分析 FlinkKafkaProducer 源码中是如何实现两阶段提交的，并保证了在结合 Kafka 的时候做到端到端的 Exactly Once 语义的。

https://flink.apache.org/features/2018/03/01/end-to-end-exactly-once-apache-flink.html

TwoPhaseCommitSinkFunction 分析

public abstract class TwoPhaseCommitSinkFunction<IN, TXN, CONTEXT>

TwoPhaseCommitSinkFunction 实现了 CheckpointedFunction 和 CheckpointListener 接口，首先就是在 initializeState 方法中开启事务，对于 Flink sink 的两阶段提交，第一阶段就是执行 CheckpointedFunction#snapshotState 当所有 task 的 checkpoint 都完成之后，每个 task 会执行 CheckpointedFunction#notifyCheckpointComplete 也就是所谓的第二阶段。

■ FlinkKafkaProducer 第一阶段分析

Flink Kafka 端到端 Exactly

@Override

这部分代码的核心在于：

先执行 preCommit 方法，EXACTLY_ONCE 模式下会调 flush，立即将数据发送到指定的 topic，这时如果消费这个 topic，需要指定 isolation.level 为 read_committed 表示消费端应用不可以看到未提交的事物内的消息。

@Override

注意第一次调用的 send 和 flush 的事务都是在 initializeState 方法中开启事务。

transaction.producer.send(record, callback);


 





 






  




transaction.producer.flush();

pendingCommitTransactions 保存了每个 checkpoint 对应的事务，并为下一次 checkpoint 创建新的 producer 事务，即 currentTransactionHolder = beginTransactionInternal();下一次的 send 和 flush 都会在这个事务中。也就是说第一阶段每一个 checkpoint 都有自己的事务，并保存在 pendingCommitTransactions 中。

■ FlinkKafkaProducer 第二阶段分析

Flink Kafka 端到端 Exactly

当所有 checkpoint 都完成后，会进入第二阶段的提交。

@Override

这一阶段会将 pendingCommitTransactions 中的事务全部提交。

@Override

这时消费端就能看到 read_committed 的数据了，至此整个 producer 的流程全部结束。

■ Exactly-Once 分析

当输入源和输出都是 kafka 的时候，Flink 之所以能做到端到端的 Exactly-Once 语义，主要是因为第一阶段 FlinkKafkaConsumer 会将消费的 offset 信息通过checkpoint 保存，所有 checkpoint 都成功之后，第二阶段 FlinkKafkaProducer 才会提交事务，结束 producer 的流程。这个过程中很大程度依赖了 kafka producer 事务的机制。

Flink 中 notifyCheckpointComplete 方法调用顺序

定义

notifyCheckpointComplete 方法在 CheckpointListener 接口中定义。

/**

简单说这个方法的含义就是在 checkpoint 做完之后，JobMaster 会通知 task 执行这个方法，例如在 FlinkKafkaProducer 中 notifyCheckpointComplete 中做了事务的提交。

样例

下面的程序会被分为两个 task，task1 是 Source: Example Source 和 task2 是 Map -> Sink: Example Sink。

DataStream<KafkaEvent> input = env.addSource(

■ operator 调用 notifyCheckpointComplete

根据上面的例子，task1 中只有一个 source 的 operator，但是 task2 中有两个operator，分别是 map 和 sink。

在 StreamTask 中，调用 task 的 notifyCheckpointComplete 方法。

@Override

其中关键的部分就是：

for (StreamOperator<?> operator : operatorChain.getAllOperators()) {

operator 的调用顺序取决于 allOperators 变量，可以看到源码中的注释，operator 是以逆序存放的。

/**

也就是说上面客户端的代码，虽然先调用了 map 后调用的 sink，但是实际执行的时候，确实先调用 sink 的 notifyCheckpointComplete 方法，后调用 map 的。

对 Exactly-Once 语义的影响

上面的例子，是先执行 source 的 notifyCheckpointComplete 方法，再执行 sink 的 notifyCheckpointComplete 方法。但是如果把 .keyBy("word") 去掉，那么只会有一个 task，所有 operator 逆序执行，也就是先调用 sink 的 notifyCheckpointComplete 方法再调用 source 的。

为了方便理解整个流程，下文只考察并发度为1的情况，不考虑部分 subtask 成功部分不成功的情况。

Tips： 以下讨论的都是基于 kafka source 和 sink

■ 先 sink 后 source

sink 成功之后 source 执行之前

sink 成功之前

checkpoint 恢复

exactly-once

__consumer_offsets 恢复

重复消费

sink 成功之后 source 执行之前，表示 sink 的 notifyCheckpointComplete 方法执行成功了，但是在执行 source 的 notifyCheckpointComplete 方法之前任务失败。

sink 成功之前，表示 sink 的 notifyCheckpointComplete 方法执行失败，提交事务失败。

测试用例

测试代码主体架构如下：

DataStream<KafkaEvent> input = env.addSource(

测试环境采用的是 Flink 1.9.0 Standalone Cluster 模式，一个 JobManager，一个TaskManager，默认只保存一个 checkpoint。

模拟异常的方法，通过 kill -9 杀掉 JobManager 和 TaskManager 进程。

在 FlinkKafkaProducer#commit 方法第一行设置断点，当程序走到这个断点的时候 kill -9 杀掉 JobManager 和 TaskManager 进程，模拟 sink 的notifyCheckpointComplete 方法执行失败的场景；
监控1，通过 bin/kafka-console-consumer.sh --topic bar --bootstrap-server 10.1.236.66:9092 监控 producer 是否 flush 数据；监控2，通过 bin/kafka-console-consumer.sh --topic bar --bootstrap-server 10.1.236.66:9092 --isolation-level read_committed 监控 producer 的事务是否成功提交；监控3，通过 bin/kafka-console-consumer.sh --topic __consumer_offsets --bootstrap-server 10.1.236.66:9092 --formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter" --consumer.config /tmp/Consumer.properties 监控 consumer 的offset 是否提交到 kafka；
发送数据一条数据 a,5,1572845161023，当走到断点的时候，说明 consumer 的 checkpoint 已经生成，但是还没有将 offset 提交到 kafka，也就是checkpoint 认为 offset 已经成功发送，但是 kafka 认为并没有发送，监控1有数据，监控2和监控3都没有数据。kill -9 杀掉 JobManager 和 TaskManager进程；
重新启动，并提交作业，不指定 checkpoint 路径。监控1，2，3，都有数据，所以这种情况，监控2，只收到了一次数据，也就是 exactly-once。这时候监控3收到的数据为：partition0 的 offset=37，partition1 的 offset=43，partition2 的 offset=39；
同样1-3步骤，发送数据一条数据 b,6,1572845161023，第4步，启动作业的时候通过-s指定要恢复的 checkpoint 路径，启动后监控1，2都没有数据，但是监控3的数据为：partition0 的 offset=37，partition1 的 offset=43，partition2 的 offset=40，再查看 task 的日志 FlinkKafkaConsumerBase - Consumer subtask 0 restored state: {KafkaTopicPartition{topic='foo', partition=0}=36, KafkaTopicPartition{topic='foo', partition=1}=42, KafkaTopicPartition{topic='foo', partition=2}=39}.，说明 checkpoint 认为上一次 partition2 的 offset=39 已经成功消费，所以恢复之后向 kafka 发送的offset 为 40。这样就导致了 partition2 的 offset=39 这条数据丢失。

同样的方法可以测试 sink 成功之后 source 执行之前的场景，只是这时候需要将断点设置在 TwoPhaseCommitSinkFunction#notifyCheckpointComplete 方法的最后一行，这样就会发现故障之前，监控1，2都是有数据的，监控3没有数据。不指定 checkpoint 路径恢复，监控1，2都会收到数据，这样就导致了重复消费。如果指定 checkpoint 路径消费，那么监控1，2就不会收到数据，保证了 exactly-once。

原因分析

产生上面情况的原因主要就是因为 checkpoint 存储的 offset 和 kafka 中的 offset 不一致导致的。

■ 先 source 后 sink

需要说明的一点这个场景的两个 task 实际是并行的，并没有绝对的先后关系，只是会有这种前后关系的可能。

source 成功之后 sink 执行之前

source 成功之前

checkpoint 恢复

丢数据

__consumer_offsets 恢复

丢数据

source 成功之后 sink 执行之前，表示 source 的 notifyCheckpointComplete 方法执行成功了，但是在执行 sink 的 notifyCheckpointComplete 方法之前任务失败。

source 成功之前，表示 source 的 notifyCheckpointComplete 方法执行失败，提交事务失败。

测试用例

模拟 source 成功之后 sink 执行之前：

需要在上面的用例中加入 keyby 算子，确保生成两个 task，监控3收到数据的时候说明 consumer 的 notifyCheckpointComplete 方法已经执行完。在FlinkKafkaProducer#commit 方法第一行设置断点，当程序走到这个断点并且监控3收到数据的时候，kill -9 杀掉 JobManager 和 TaskManager 进程，模拟 sink 执行 notifyCheckpointComplete 方法失败的场景；
这时候重启作业，checkpoint 和 kafka 中 offset 已经是一致的了，无论是从checkpoint 还是 kafka，都是一样的。所以 source 认为已经成功消费了，不会再读上次的 offset，都会导致数据丢失。

source 成功之前：

对于在 source 之前程序就挂掉，相当于所有的 operator 都没有执行notifyCheckpointComplete 方法，但是 source 的 checkpoint 已经做过了，只是没有将 offset 发送到 kafka，这样只有从 __consumer_offsets 恢复才能保证不丢数据。

小结：本节通过一种极端的测试场景希望让读者可以更深入的理解 Flink 中的 Exactly-Once 语义。在程序挂了以后需要排查是什么原因和什么阶段导致的，才能通过合适的方式恢复作业。在实际的生产环境中，会有重试或者更多的方式保证高可用，也建议保留多个 checkpoint，以便业务上可以恢复正确的数据。

作者介绍：

吴鹏，亚信科技资深工程师，Apache Flink Contributor。先后就职于中兴，IBM，华为。目前在亚信科技负责实时流处理引擎产品的研发。

▼ 更多技术文章 ▼