查看原文
其他

Kafka设计-恰好一次和事务消息

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/
欢迎点赞+收藏+转发朋友圈素质三连


文章不错?点个【在看】吧! 👇

: . Video Mini Program Like ,轻点两下取消赞 Wow ,轻点两下取消在看

您可能也对以下帖子感兴趣

文章有问题?点此查看未经处理的缓存