阿里面试，Flink向Kafka中写入数据，如何保证数据的一致性

　　这是一个朋友半个月前，去阿里面试的面试题。关于这道题的答案，我已经录制成了视频，欢迎朋友们观看。不喜欢看视频的朋友，也可以看看这篇文章。
　　阿里面试，Flink向Kafka中写数据，是如何保证数据一致性的FlinkKafkaProducer的事务一致性
　　我们都知道Kafka是非常受欢迎的分布式消息系统，在Flink应用中，Kafka既可以作为Source，同时也可以作为Sink。从Kafka0.11.0及以上版本就提供了对事务的支持，而Flink从Flink1.7及以上版本，就通过FlinkKafkaProducer实现了两阶段事务提交的能力。
　　FlinkKafkaProducer作为Kafka的Sink，之所以能够实现两阶段提交，主要是它继承了抽象类TwoPhaseCommitSinkFunction，根据这个名字就能知道这个抽象类实现了两阶段提交。 @PublicEvolving public class FlinkKafkaProducer         extends TwoPhaseCommitSinkFunction<                 IN,                 FlinkKafkaProducer.KafkaTransactionState,                 FlinkKafkaProducer.KafkaTransactionContext> { }
　　FlinkKafkaProducer要实现一致性，必然离不开TwoPhaseCommitSinkFunction实现了两个接口，一个是 CheckpointedFunction，另外一个是CheckpointListener。@PublicEvolving public abstract class TwoPhaseCommitSinkFunction extends RichSinkFunction         implements CheckpointedFunction, CheckpointListener { }
　　下面我们具体看看TwoPhaseCommitSinkFunction是如何工作的。TwoPhaseCommitSinkFunction
　　TwoPhaseCommitSinkFunction保留了5个函数需要子类去实现， invoke()
　　FlinkKafkaProducer在接收到每一条一条消息时，都会调用invoke()方法，在invoke()方法中，会将输入的消息（next）封装成ProducerRecord，然后调用Kafka客户端的KafkaProducer.send方法，将ProducerRecord 对象写入缓冲。public void invoke(             FlinkKafkaProducer.KafkaTransactionState transaction, IN next, Context context)             throws FlinkKafkaException {         checkErroneous();          ProducerRecord record;         if (keyedSchema != null) {             byte[] serializedKey = keyedSchema.serializeKey(next);             byte[] serializedValue = keyedSchema.serializeValue(next);             String targetTopic = keyedSchema.getTargetTopic(next);             if (targetTopic == null) {                 targetTopic = defaultTopicId;             }              Long timestamp = null;             if (this.writeTimestampToKafka) {                 timestamp = context.timestamp();             }              int[] partitions = topicPartitionsMap.get(targetTopic);             if (null == partitions) {                 partitions = getPartitionsByTopic(targetTopic, transaction.producer);                 topicPartitionsMap.put(targetTopic, partitions);             }             if (flinkKafkaPartitioner != null) {                 record =                         new ProducerRecord<>(                                 targetTopic,                                 flinkKafkaPartitioner.partition(                                         next,                                         serializedKey,                                         serializedValue,                                         targetTopic,                                         partitions),                                 timestamp,                                 serializedKey,                                 serializedValue);             } else {                 record =                         new ProducerRecord<>(                                 targetTopic, null, timestamp, serializedKey, serializedValue);             }         } else if (kafkaSchema != null) {             if (kafkaSchema instanceof KafkaContextAware) {                 @SuppressWarnings(＂unchecked＂)                 KafkaContextAware contextAwareSchema = (KafkaContextAware) kafkaSchema;                  String targetTopic = contextAwareSchema.getTargetTopic(next);                 if (targetTopic == null) {                     targetTopic = defaultTopicId;                 }                 int[] partitions = topicPartitionsMap.get(targetTopic);                  if (null == partitions) {                     partitions = getPartitionsByTopic(targetTopic, transaction.producer);                     topicPartitionsMap.put(targetTopic, partitions);                 }                  contextAwareSchema.setPartitions(partitions);             }             record = kafkaSchema.serialize(next, context.timestamp());         } else {             throw new RuntimeException(                     ＂We have neither KafkaSerializationSchema nor KeyedSerializationSchema, this＂                             + ＂is a bug.＂);         }          pendingRecords.incrementAndGet();         transaction.producer.send(record, callback);     }
　　数据写入缓冲池beginTransaction()、preCommit()、commit()和abort()这四个方法，定义了两阶段提交协议的几个步骤。对于Kafka这种外部系统来说，就是通过调用两阶段提交协议对于的函数实现数据写入功能。状态快照处理
　　一 在preCommit()函数中，调用 KafkaProducer 客户端 flush 方法，将缓冲区内全部记录发送到 Kafka， 但不提交 。
　　TwoPhaseCommitSinkFunction里面 的preCommit()函数是在snapshotState方法中调用的，而snapshotState方法是在算子Checkpoint的时候触发的。这样就保证了算子在做Checkpoint时，所有该Checkpoint之前的数据都已经安全的发送到了下游（而不是在缓存中），如下图所示。 这些记录会写到Kafka的Topic-A中，此时数据的隔离级别是UNCOMMITED，如果有一个Topic-A的消费者程序，正在消费这个Topic中的数据，并且设置的 isolation.level=read_committed，那么此时这个消费端还无法 poll 到 flush 的数据，因为这些数据尚未 commit。具体什么时候commit呢？ 在快照结束处理阶段会commit，我们稍后会讲到这一点。
　　Flush数据到Kafka
　　二 将本轮的CheckPointId与当前事务记录到一个 Map 表示的待提交事务集合中。
　　在Flush完数据之后，Flink会将CheckPointID与当前事务记录到一个Map中，这个Map表示的待提交事务集合，key 是当前快照的 CheckpointId，value 是由 TransactionHolder 表示的事务对象。TransactionHolder 对象内部记录了 transactionalId、producerId、epoch 以及 Kafka 客户端 kafkaProducer 的引用。
　　记录到Map中
　　那么，这个TransactionID是怎么产生的呢？Flink用一个队列作为transactional id的Pool，新的Transaction开始时从队头拿出一个transactional id，Transaction结束时将transactional id放回队尾。 private FlinkKafkaInternalProducer createTransactionalProducer()             throws FlinkKafkaException {         String transactionalId = availableTransactionalIds.poll();         if (transactionalId == null) {             throw new FlinkKafkaException(                     FlinkKafkaErrorCode.PRODUCERS_POOL_EMPTY,                     ＂Too many ongoing snapshots. Increase kafka producers pool size or decrease number of concurrent checkpoints.＂);         }         FlinkKafkaInternalProducer producer =                 initTransactionalProducer(transactionalId, true);         producer.initTransactions();         return producer;     }  private void recycleTransactionalProducer(FlinkKafkaInternalProducer producer) {         availableTransactionalIds.add(producer.getTransactionalId());         producer.flush();         producer.close(Duration.ofSeconds(0));     }
　　三 持久化当前事务处理状态，也就是将当前处理的事务详情存入状态后端，供应用恢复时使用
　　快照持久化快照结束处理
　　TwoPhaseCommitSinkFunction 实现了 CheckpointListener，应用中所有算子的快照处理成功后会收到基于某轮 Checkpoint 完成的通知。
　　FlinkKafkaProducer#commit()方法，这个方法实际上是代理了KafkaProducer.commitTransaction()方法，正式向Kafka提交事务。 protected void commit(FlinkKafkaProducer.KafkaTransactionState transaction) {         if (transaction.isTransactional()) {             try {                 transaction.producer.commitTransaction();             } finally {                 recycleTransactionalProducer(transaction.producer);             }         }     }
　　该方法的调用点位于TwoPhaseCommitSinkFunction.notifyCheckpointComplete()方法中。顾名思义，当所有检查点都成功完成之后，会回调这个方法。
　　该方法每次从正在等待提交的事务句柄中取出一个，校验它的检查点ID，并调用commit()方法提交。
　　根据CheckPointId的值，到＂待提交事务中＂，获取到事务Id之后，就可以提交事务了。而具体要提交哪些事务，是从上一阶段生成的待提交事务集合中获取到的。
　　在下图中的第⑫步执行成功后，flush 到 Kafka 的数据从 UNCOMMITTED 变更为 COMMITTED，这意味着此时消费端可以 poll 到这批数据了。
　　提交事务总结
　　Flink借用两阶段提交原理，在加上Kafka的事务特性实现了，Flink向Kafka发送数据的端到端Exactly_Once。
　　而（2PC）两阶段提交的第一个阶段就是通过Kafka的事务初始化、事务开启、flush()等操作进行预提交事务；第二个阶段提交就是通过Kafka的commit操作来真正执行事务提交。
　　在了解了两阶段提交原理之后，可以很快的将这种方案套用到，其他支持事务的外部存储或消息队列中。
　　希望这篇文章能起到抛砖引玉的作用，欢迎大家评论、转发！