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: 正常消息；
new_seq<=old_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所示，
图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中的消息就丢失了。
2）假如consumer只消费了topic_a，没有消费topic_b，这样也不能读到完整的事务消息。
3）典型的kafka stream应用从多个topic消费，然后向一个或多个topic写。在一次故障后，kafka stream应用重新开始处理流数据，由于从多个topic读到的数据之间不存在稳定的顺序(即便只有一个topic，从多个分区读到的数据之间也没有稳定的顺序)，那么两次处理输出的结果就可能会不一样。
也就是说，虽然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/
