1.幂等消息

为了解决重试导致的消息重复、乱序问题，kafka引入了幂等消息。幂等消息保证producer在一次会话内写入一个partition内的消息具有幂等性，可以通过重试来确保消息发布的Exactly Once语义。

实现逻辑很简单:

区分producer会话

producer每次启动后，首先向broker申请一个全局唯一的pid，用来标识本次会话。

消息检测

message_v2 增加了sequence number字段，producer每发一批消息，seq就加1。

broker在内存维护(pid,seq)映射，收到消息后检查seq，如果，

new_seq=old_seq+1: 正常消息；

producer重试

producer在收到明确的的消息丢失ack，或者超时后未收到ack，要进行重试。

2.事务消息

考虑在stream处理的场景中，需要多个消息的原子写入语义，要么全部写入成功，要么全部失败，这就是kafka事务消息要解决的问题。

事务消息是由producer、事务协调器、broker、组协调器、consumer共同参与实现的，

1）producer

为producer指定固定的TransactionalId，可以穿越producer的多次会话(producer重启/断线重连)中，持续标识producer的身份。

使用epoch标识producer的每一次"重生"，防止同一producer存在多个会话。

producer遵从幂等消息的行为，并在发送的recordbatch中增加事务id和epoch。

2）事务协调器(Transaction Coordinator)

引入事务协调器，以两阶段提交的方式，实现消息的事务提交。

事务协调器使用一个特殊的topic：transaction，来做事务提交日志。

事务控制器通过RPC调研，协调 broker 和 consumer coordinator 实现事务的两阶段提交。

每一个broker都会启动一个事务协调器，使用hash(TransactionalId)确定producer对应的事务协调器，使得整个集群的负载均衡。

3) broker

broker处理在事务协调器的commit/abort控制消息，把控制消息向正常消息一样写入topic(和正常消息交织在一起，用来确认事务提交的日志偏移)，并向前推进消息提交偏移hw。

4) 组协调器

如果在事务过程中，提交了消费偏移，组协调器在offset log中写入事务消费偏移。当事务提交时，在offset log中写入事务offset确认消息。

5）consumer

consumer过滤未提交消息和事务控制消息，使这些消息对用户不可见。

有两种实现方式:

consumer缓存方式

设置isolation.level=read_uncommitted，此时topic的所有消息对consumer都可见。

consumer缓存这些消息，直到收到事务控制消息。若事务commit，则对外发布这些消息；若事务abort，则丢弃这些消息。

broker过滤方式

设置isolation.level=read_committed，此时topic中未提交的消息对consumer不可见，只有在事务结束后，消息才对consumer可见。

broker给consumer的BatchRecord消息中，会包含以列表，指明哪些是"abort"事务，consumer丢弃abort事务的消息即可。

事务消息处理流程如图1所示，

Kafka设计

图1 事务消息业务流程

流程说明：

1. 查找事务协调器 -- FindCoordinatorRequest

事务协调器是分配pid和管理事务的核心，produer首先对任何一个broker发送FindCoordinatorRequest，发现自己的事务协调器。

2. 申请pid -- InitPidRequest

紧接着，producer向事务协调器发送InitPidRequest，申请生成pid。

2a.当指定了transactional.id时，事务协调器为producer分区pid，并更新epoch，把(tid,pid)的映射关系写入事务日志。同时清理tid任何未完成的事务，丢弃未提交的消息。

3. 启动事务

启动事务是producer的本地操作，促使producer更新内部状态，不会和事务协调器发生关系。

事务协调器自动启动事务，始终处在一个接一个的事务处理状态机中。

4. consume-transform-produce 事务循环

4.1. 注册partition -- AddPartitionsToTxnRequest

对于每一个要在事务中写消息的topic分区，producer应当在第一次发消息前，向事务处理器注册分区。

4.1a.事务处理器把事务关联的分区写入事务日志。

在提交或终止事务时，事务协调器需要这些信息，控制事务涉及的所有分区leader完成事务提交或终止。

4.2. 写消息 -- ProduceRequest

4.2a. producer向分区leader写消息，消息中包含tid,pid,epoch和seq。

4.3. 提交消费偏移 -- AddOffsetCommitsToTxnRequest

4.3a. producer向事务协调器发送消费偏移，事务协调器在事务日志中记录偏移信息，并把组协调器返回给producer。

4.4. 提交消费偏移 -- TxnOffsetCommitRequest

4.4a. producer向组协调器发送TxnOffsetCommitRequest，组协调器把偏移信息写入偏移日志。但是，要一直等到事务提交后，这个偏移才生效，对外部可见。

5. 提交或终止事务

5.1. EndTxnRequest

收到提交或终止事务的请求时，事务处理器执行下面的操作：

1. 在事务日志中写入PREPARE_COMMIT或PREPARE_ABORT消息(5.1a)。

2. 通过WriteTxnMarkerRequest向事务中的所有broker发事务控制消息(5.2)。

3. 在事务之日中写入COMMITTED或ABORTED消息(5.3)。

5.2. WriteTxnMarkerRequest

这个消息由事务处理器发给事务中所涉及分区的leader。

当收到这个消息后，broker会在分区log中写入一个COMMIT或ABORT控制消息。同时，也会更新该分区的事务提交偏移hw。

如果事务中有提交消费偏移， broker也会把控制消息写入 __consumer-offsets log，并通知组协调器使事务中提交的消费偏移生效。

5.3. 写最终的commit或abort消息

当所有的commit或abort消息写入数据日志，事务协调器在事务日志中写入事务日志，标志这事务结束。

至此，本事务的所有状态信息都可以被删除，可以开始一个新的事务。

在实现上，还有很多细节，比如，事务协调器会启动定时器，用来检测并终止开始后长时间不活动的事务，具体请参考下面列出的kafka社区技术文档。

【总结】：

我们要认识到，虽然kafka事务消息提供了多个消息原子写的保证，但它不保证原子读。

例如，

1）事务向topic_a和topic_b两个分区写入消息，在事务提交后的某个时刻，topic_a的全部副本失效。这时topic_b中的消息可以正常消费，但topic_a中的消息就丢失了。

也就是说，虽然kafka log持久化了数据，也可以通过指定offset多次消费数据，但由于分区数据之间的无序性，导致每次处理输出的结果都是不同的。这使得kafka stream不能像hadoop批处理任务一样，可以随时重新执行，保证每次执行的结果相同。除非我们只从一个topic分区读数据。

【参考】：

[0] https://cwiki.apache.org/confluence/display/KAFKA/Idempotent+Producer
[1]https://cwiki.apache.org/confluence/display/KAFKA/KIP-98+-+Exactly+Once+Delivery+and+Transactional+Messaging#KIP-98-ExactlyOnceDeliveryandTransactionalMessaging-4.1AddPartitionsToTxnRequest
[2]https://docs.google.com/document/d/11Jqy\_GjUGtdXJK94XGsEIK7CP1SnQGdp2eF0wSw9ra8/edit#
[3]https://cwiki.apache.org/confluence/display/KAFKA/Transactional+Messaging+in+Kafka
[4]https://www.confluent.io/blog/exactly-once-semantics-are-possible-heres-how-apache-kafka-does-it/
[5]https://www.confluent.io/blog/transactions-apache-kafka/

欢迎点赞+收藏+转发朋友圈素质三连