Kafka设计解析（八）：Kafka事务机制与Exactly Once语义实现原理-InfoQ

低代码到底是不是行业毒瘤？一线大厂怎么做的？戳此了解>>> 了解详情 



 写点什么

本文所有 Kafka 原理性的描述除特殊说明外均基于 Kafka 1.0.0 版本。

Kafka 事务机制的实现主要是为了支持

Exactly Once即正好一次语义
操作的原子性
有状态操作的可恢复性

`Exactly Once`

《 Kafka 背景及架构介绍》一文中有说明 Kafka 在 0.11.0.0 之前的版本中只支持At Least Once和At Most Once语义，尚不支持Exactly Once语义。

但是在很多要求严格的场景下，如使用 Kafka 处理交易数据，Exactly Once语义是必须的。我们可以通过让下游系统具有幂等性来配合 Kafka 的At Least Once语义来间接实现Exactly Once。但是：

该方案要求下游系统支持幂等操作，限制了 Kafka 的适用场景
实现门槛相对较高，需要用户对 Kafka 的工作机制非常了解
对于 Kafka Stream 而言，Kafka 本身即是自己的下游系统，但 Kafka 在 0.11.0.0 版本之前不具有幂等发送能力

因此，Kafka 本身对Exactly Once语义的支持就非常必要。

操作原子性

操作的原子性是指，多个操作要么全部成功要么全部失败，不存在部分成功部分失败的可能。

实现原子性操作的意义在于：

操作结果更可控，有助于提升数据一致性
便于故障恢复。因为操作是原子的，从故障中恢复时只需要重试该操作（如果原操作失败）或者直接跳过该操作（如果原操作成功），而不需要记录中间状态，更不需要针对中间状态作特殊处理

幂等性发送

上文提到，实现Exactly Once的一种方法是让下游系统具有幂等处理特性，而在 Kafka Stream 中，Kafka Producer 本身就是“下游”系统，因此如果能让 Producer 具有幂等处理特性，那就可以让 Kafka Stream 在一定程度上支持Exactly once语义。

为了实现 Producer 的幂等语义，Kafka 引入了Producer ID（即PID）和Sequence Number。每个新的 Producer 在初始化的时候会被分配一个唯一的 PID，该 PID 对用户完全透明而不会暴露给用户。

对于每个 PID，该 Producer 发送数据的每个<Topic, Partition>都对应一个从 0 开始单调递增的Sequence Number。

类似地，Broker 端也会为每个<PID, Topic, Partition>维护一个序号，并且每次 Commit 一条消息时将其对应序号递增。对于接收的每条消息，如果其序号比 Broker 维护的序号（即最后一次 Commit 的消息的序号）大一，则 Broker 会接受它，否则将其丢弃：

如果消息序号比 Broker 维护的序号大一以上，说明中间有数据尚未写入，也即乱序，此时 Broker 拒绝该消息，Producer 抛出InvalidSequenceNumber
如果消息序号小于等于 Broker 维护的序号，说明该消息已被保存，即为重复消息，Broker 直接丢弃该消息，Producer 抛出DuplicateSequenceNumber

上述设计解决了 0.11.0.0 之前版本中的两个问题：

Broker 保存消息后，发送 ACK 前宕机，Producer 认为消息未发送成功并重试，造成数据重复
前一条消息发送失败，后一条消息发送成功，前一条消息重试后成功，造成数据乱序

事务性保证

上述幂等设计只能保证单个 Producer 对于同一个<Topic, Partition>的Exactly Once语义。

另外，它并不能保证写操作的原子性——即多个写操作，要么全部被 Commit 要么全部不被 Commit。

更不能保证多个读写操作的的原子性。尤其对于 Kafka Stream 应用而言，典型的操作即是从某个 Topic 消费数据，经过一系列转换后写回另一个 Topic，保证从源 Topic 的读取与向目标 Topic 的写入的原子性有助于从故障中恢复。

事务保证可使得应用程序将生产数据和消费数据当作一个原子单元来处理，要么全部成功，要么全部失败，即使该生产或消费跨多个<Topic, Partition>。

另外，有状态的应用也可以保证重启后从断点处继续处理，也即事务恢复。

为了实现这种效果，应用程序必须提供一个稳定的（重启后不变）唯一的 ID，也即Transaction ID。Transactin ID与PID可能一一对应。区别在于Transaction ID由用户提供，而PID是内部的实现对用户透明。

另外，为了保证新的 Producer 启动后，旧的具有相同Transaction ID的 Producer 即失效，每次 Producer 通过Transaction ID拿到 PID 的同时，还会获取一个单调递增的 epoch。由于旧的 Producer 的 epoch 比新 Producer 的 epoch 小，Kafka 可以很容易识别出该 Producer 是老的 Producer 并拒绝其请求。

有了Transaction ID后，Kafka 可保证：

跨 Session 的数据幂等发送。当具有相同Transaction ID的新的 Producer 实例被创建且工作时，旧的且拥有相同Transaction ID的 Producer 将不再工作。
跨 Session 的事务恢复。如果某个应用实例宕机，新的实例可以保证任何未完成的旧的事务要么 Commit 要么 Abort，使得新实例从一个正常状态开始工作。

需要注意的是，上述的事务保证是从 Producer 的角度去考虑的。从 Consumer 的角度来看，该保证会相对弱一些。尤其是不能保证所有被某事务 Commit 过的所有消息都被一起消费，因为：

对于压缩的 Topic 而言，同一事务的某些消息可能被其它版本覆盖
事务包含的消息可能分布在多个 Segment 中（即使在同一个 Partition 内），当老的 Segment 被删除时，该事务的部分数据可能会丢失
Consumer 在一个事务内可能通过 seek 方法访问任意 Offset 的消息，从而可能丢失部分消息
Consumer 可能并不需要消费某一事务内的所有 Partition，因此它将永远不会读取组成该事务的所有消息

事务性消息传递

这一节所说的事务主要指原子性，也即 Producer 将多条消息作为一个事务批量发送，要么全部成功要么全部失败。

为了实现这一点，Kafka 0.11.0.0 引入了一个服务器端的模块，名为Transaction Coordinator，用于管理 Producer 发送的消息的事务性。

该Transaction Coordinator维护Transaction Log，该 log 存于一个内部的 Topic 内。由于 Topic 数据具有持久性，因此事务的状态也具有持久性。

Producer 并不直接读写Transaction Log，它与Transaction Coordinator通信，然后由Transaction Coordinator将该事务的状态插入相应的Transaction Log。

Transaction Log的设计与Offset Log用于保存 Consumer 的 Offset 类似。

事务中 Offset 的提交

许多基于 Kafka 的应用，尤其是 Kafka Stream 应用中同时包含 Consumer 和 Producer，前者负责从 Kafka 中获取消息，后者负责将处理完的数据写回 Kafka 的其它 Topic 中。

为了实现该场景下的事务的原子性，Kafka 需要保证对 Consumer Offset 的 Commit 与 Producer 对发送消息的 Commit 包含在同一个事务中。否则，如果在二者 Commit 中间发生异常，根据二者 Commit 的顺序可能会造成数据丢失和数据重复：

如果先 Commit Producer 发送数据的事务再 Commit Consumer 的 Offset，即At Least Once语义，可能造成数据重复。
如果先 Commit Consumer 的 Offset，再 Commit Producer 数据发送事务，即At Most Once语义，可能造成数据丢失。

用于事务特性的控制型消息

为了区分写入 Partition 的消息被 Commit 还是 Abort，Kafka 引入了一种特殊类型的消息，即Control Message。该类消息的 Value 内不包含任何应用相关的数据，并且不会暴露给应用程序。它只用于 Broker 与 Client 间的内部通信。

对于 Producer 端事务，Kafka 以 Control Message 的形式引入一系列的Transaction Marker。Consumer 即可通过该标记判定对应的消息被 Commit 了还是 Abort 了，然后结合该 Consumer 配置的隔离级别决定是否应该将该消息返回给应用程序。

事务处理样例代码

复制代码

 Producer<String, String> producer = new KafkaProducer<String, String>(props);
 
// 初始化事务，包括结束该 Transaction ID 对应的未完成的事务（如果有）
// 保证新的事务在一个正确的状态下启动
producer.initTransactions();
 
// 开始事务
producer.beginTransaction();
 
// 消费数据
ConsumerRecords<String, String> records = consumer.poll(100);
 
try{
    // 发送数据
    producer.send(new ProducerRecord<String, String>("Topic", "Key", "Value"));
 
    // 发送消费数据的 Offset，将上述数据消费与数据发送纳入同一个 Transaction 内
    producer.sendOffsetsToTransaction(offsets, "group1");
 
    // 数据发送及 Offset 发送均成功的情况下，提交事务
    producer.commitTransaction();
} catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
    // 数据发送或者 Offset 发送出现异常时，终止事务
    producer.abortTransaction();
} finally {
    // 关闭 Producer 和 Consumer
    producer.close();
    consumer.close();
}

完整事务过程

找到`Transaction Coordinator`

由于Transaction Coordinator是分配 PID 和管理事务的核心，因此 Producer 要做的第一件事情就是通过向任意一个 Broker 发送FindCoordinator请求找到Transaction Coordinator的位置。

注意：只有应用程序为 Producer 配置了Transaction ID时才可使用事务特性，也才需要这一步。另外，由于事务性要求 Producer 开启幂等特性，因此通过将transactional.id设置为非空从而开启事务特性的同时也需要通过将enable.idempotence设置为 true 来开启幂等特性。

获取 PID

找到Transaction Coordinator后，具有幂等特性的 Producer 必须发起InitPidRequest请求以获取 PID。

注意：只要开启了幂等特性即必须执行该操作，而无须考虑该 Producer 是否开启了事务特性。

* 如果事务特性被开启 *
InitPidRequest会发送给Transaction Coordinator。如果Transaction Coordinator是第一次收到包含有该Transaction ID的 InitPidRequest 请求，它将会把该<TransactionID, PID>存入Transaction Log，如上图中步骤 2.1 所示。这样可保证该对应关系被持久化，从而保证即使Transaction Coordinator宕机该对应关系也不会丢失。

除了返回 PID 外，InitPidRequest还会执行如下任务：

增加该 PID 对应的 epoch。具有相同 PID 但 epoch 小于该 epoch 的其它 Producer（如果有）新开启的事务将被拒绝。
恢复（Commit 或 Abort）之前的 Producer 未完成的事务（如果有）。

注意：InitPidRequest的处理过程是同步阻塞的。一旦该调用正确返回，Producer 即可开始新的事务。

另外，如果事务特性未开启，InitPidRequest可发送至任意 Broker，并且会得到一个全新的唯一的 PID。该 Producer 将只能使用幂等特性以及单一 Session 内的事务特性，而不能使用跨 Session 的事务特性。

开启事务

Kafka 从 0.11.0.0 版本开始，提供beginTransaction()方法用于开启一个事务。调用该方法后，Producer 本地会记录已经开启了事务，但Transaction Coordinator只有在 Producer 发送第一条消息后才认为事务已经开启。

Consume-Transform-Produce

这一阶段，包含了整个事务的数据处理过程，并且包含了多种请求。

AddPartitionsToTxnRequest
一个 Producer 可能会给多个<Topic, Partition>发送数据，给一个新的<Topic, Partition>发送数据前，它需要先向Transaction Coordinator发送AddPartitionsToTxnRequest。

Transaction Coordinator会将该<Transaction, Topic, Partition>存于Transaction Log内，并将其状态置为BEGIN，如上图中步骤 4.1 所示。有了该信息后，我们才可以在后续步骤中为每个Topic, Partition>设置 COMMIT 或者 ABORT 标记（如上图中步骤 5.2 所示）。

另外，如果该<Topic, Partition>为该事务中第一个<Topic, Partition>，Transaction Coordinator还会启动对该事务的计时（每个事务都有自己的超时时间）。

ProduceRequest
Producer 通过一个或多个ProduceRequest发送一系列消息。除了应用数据外，该请求还包含了 PID，epoch，和Sequence Number。该过程如上图中步骤 4.2 所示。

AddOffsetsToTxnRequest
为了提供事务性，Producer 新增了sendOffsetsToTransaction方法，该方法将多组消息的发送和消费放入同一批处理内。

该方法先判断在当前事务中该方法是否已经被调用并传入了相同的 Group ID。若是，直接跳到下一步；若不是，则向Transaction Coordinator发送AddOffsetsToTxnRequests请求，Transaction Coordinator将对应的所有<Topic, Partition>存于Transaction Log中，并将其状态记为BEGIN，如上图中步骤 4.3 所示。该方法会阻塞直到收到响应。

TxnOffsetCommitRequest
作为sendOffsetsToTransaction方法的一部分，在处理完AddOffsetsToTxnRequest后，Producer 也会发送TxnOffsetCommit请求给Consumer Coordinator从而将本事务包含的与读操作相关的各<Topic, Partition>的 Offset 持久化到内部的__consumer_offsets中，如上图步骤 4.4 所示。

在此过程中，Consumer Coordinator会通过 PID 和对应的 epoch 来验证是否应该允许该 Producer 的该请求。

这里需要注意：

写入__consumer_offsets的 Offset 信息在当前事务 Commit 前对外是不可见的。也即在当前事务被 Commit 前，可认为该 Offset 尚未 Commit，也即对应的消息尚未被完成处理。
Consumer Coordinator并不会立即更新缓存中相应<Topic, Partition>的 Offset，因为此时这些更新操作尚未被 COMMIT 或 ABORT。

Commit 或 Abort 事务

一旦上述数据写入操作完成，应用程序必须调用KafkaProducer的commitTransaction方法或者abortTransaction方法以结束当前事务。

EndTxnRequest
commitTransaction方法使得 Producer 写入的数据对下游 Consumer 可见。abortTransaction方法通过Transaction Marker将 Producer 写入的数据标记为Aborted状态。下游的 Consumer 如果将isolation.level设置为READ_COMMITTED，则它读到被 Abort 的消息后直接将其丢弃而不会返回给客户程序，也即被 Abort 的消息对应用程序不可见。

无论是 Commit 还是 Abort，Producer 都会发送EndTxnRequest请求给Transaction Coordinator，并通过标志位标识是应该 Commit 还是 Abort。

收到该请求后，Transaction Coordinator会进行如下操作

将PREPARE_COMMIT或PREPARE_ABORT消息写入Transaction Log，如上图中步骤 5.1 所示
通过WriteTxnMarker请求以Transaction Marker的形式将COMMIT或ABORT信息写入用户数据日志以及Offset Log中，如上图中步骤 5.2 所示
最后将COMPLETE_COMMIT或COMPLETE_ABORT信息写入Transaction Log中，如上图中步骤 5.3 所示

补充说明：对于commitTransaction方法，它会在发送EndTxnRequest之前先调用 flush 方法以确保所有发送出去的数据都得到相应的 ACK。对于abortTransaction方法，在发送EndTxnRequest之前直接将当前 Buffer 中的事务性消息（如果有）全部丢弃，但必须等待所有被发送但尚未收到 ACK 的消息发送完成。

上述第二步是实现将一组读操作与写操作作为一个事务处理的关键。因为 Producer 写入的数据 Topic 以及记录 Comsumer Offset 的 Topic 会被写入相同的Transactin Marker，所以这一组读操作与写操作要么全部 COMMIT 要么全部 ABORT。

WriteTxnMarkerRequest
上面提到的WriteTxnMarkerRequest由Transaction Coordinator发送给当前事务涉及到的每个<Topic, Partition>的 Leader。收到该请求后，对应的 Leader 会将对应的COMMIT(PID)或者ABORT(PID)控制信息写入日志，如上图中步骤 5.2 所示。

该控制消息向 Broker 以及 Consumer 表明对应 PID 的消息被 Commit 了还是被 Abort 了。

这里要注意，如果事务也涉及到__consumer_offsets，即该事务中有消费数据的操作且将该消费的 Offset 存于__consumer_offsets中，Transaction Coordinator也需要向该内部 Topic 的各 Partition 的 Leader 发送WriteTxnMarkerRequest从而写入COMMIT(PID)或COMMIT(PID)控制信息。

写入最终的COMPLETE_COMMIT或COMPLETE_ABORT消息
写完所有的Transaction Marker后，Transaction Coordinator会将最终的COMPLETE_COMMIT或COMPLETE_ABORT消息写入Transaction Log中以标明该事务结束，如上图中步骤 5.3 所示。

此时，Transaction Log中所有关于该事务的消息全部可以移除。当然，由于 Kafka 内数据是 Append Only 的，不可直接更新和删除，这里说的移除只是将其标记为 null 从而在 Log Compact 时不再保留。

另外，COMPLETE_COMMIT或COMPLETE_ABORT的写入并不需要得到所有 Rreplica 的 ACK，因为如果该消息丢失，可以根据事务协议重发。

补充说明，如果参与该事务的某些<Topic, Partition>在被写入Transaction Marker前不可用，它对READ_COMMITTED的 Consumer 不可见，但不影响其它可用<Topic, Partition>的 COMMIT 或 ABORT。在该<Topic, Partition>恢复可用后，Transaction Coordinator会重新根据PREPARE_COMMIT或PREPARE_ABORT向该<Topic, Partition>发送Transaction Marker。

总结

PID与Sequence Number的引入实现了写操作的幂等性
写操作的幂等性结合At Least Once语义实现了单一 Session 内的Exactly Once语义
Transaction Marker与PID提供了识别消息是否应该被读取的能力，从而实现了事务的隔离性
Offset 的更新标记了消息是否被读取，从而将对读操作的事务处理转换成了对写（Offset）操作的事务处理
Kafka 事务的本质是，将一组写操作（如果有）对应的消息与一组读操作（如果有）对应的 Offset 的更新进行同样的标记（即Transaction Marker）来实现事务中涉及的所有读写操作同时对外可见或同时对外不可见
Kafka 只提供对 Kafka 本身的读写操作的事务性，不提供包含外部系统的事务性

Exception 处理

InvalidProducerEpoch
这是一种 Fatal Error，它说明当前 Producer 是一个过期的实例，有Transaction ID相同但 epoch 更新的 Producer 实例被创建并使用。此时 Producer 会停止并抛出 Exception。

InvalidPidMapping
Transaction Coordinator没有与该Transaction ID对应的 PID。此时 Producer 会通过包含有Transaction ID的InitPidRequest请求创建一个新的 PID。

NotCorrdinatorForGTransactionalId
该Transaction Coordinator不负责该当前事务。Producer 会通过FindCoordinatorRequest请求重新寻找对应的Transaction Coordinator。

InvalidTxnRequest
违反了事务协议。正确的 Client 实现不应该出现这种 Exception。如果该异常发生了，用户需要检查自己的客户端实现是否有问题。

CoordinatorNotAvailable
Transaction Coordinator仍在初始化中。Producer 只需要重试即可。

DuplicateSequenceNumber
发送的消息的序号低于 Broker 预期。该异常说明该消息已经被成功处理过，Producer 可以直接忽略该异常并处理下一条消息

InvalidSequenceNumber
这是一个 Fatal Error，它说明发送的消息中的序号大于 Broker 预期。此时有两种可能

数据乱序。比如前面的消息发送失败后重试期间，新的消息被接收。正常情况下不应该出现该问题，因为当幂等发送启用时，max.inflight.requests.per.connection被强制设置为 1，而acks被强制设置为 all。故前面消息重试期间，后续消息不会被发送，也即不会发生乱序。并且只有 ISR 中所有 Replica 都 ACK，Producer 才会认为消息已经被发送，也即不存在 Broker 端数据丢失问题。
服务器由于日志被 Truncate 而造成数据丢失。此时应该停止 Producer 并将此 Fatal Error 报告给用户。

InvalidTransactionTimeout
InitPidRequest调用出现的 Fatal Error。它表明 Producer 传入的 timeout 时间不在可接受范围内，应该停止 Producer 并报告给用户。

处理`Transaction Coordinator`失败

写`PREPARE_COMMIT/PREPARE_ABORT`前失败

Producer 通过FindCoordinatorRequest找到新的Transaction Coordinator，并通过EndTxnRequest请求发起COMMIT或ABORT流程，新的Transaction Coordinator继续处理EndTxnRequest请求——写PREPARE_COMMIT或PREPARE_ABORT，写Transaction Marker，写COMPLETE_COMMIT或COMPLETE_ABORT。

写完`PREPARE_COMMIT/PREPARE_ABORT`后失败

此时旧的Transaction Coordinator可能已经成功写入部分Transaction Marker。新的Transaction Coordinator会重复这些操作，所以部分 Partition 中可能会存在重复的COMMIT或ABORT，但只要该 Producer 在此期间没有发起新的事务，这些重复的Transaction Marker就不是问题。

写完`COMPLETE_COMMIT/ABORT`后失败

旧的Transaction Coordinator可能已经写完了COMPLETE_COMMIT或COMPLETE_ABORT但在返回EndTxnRequest之前失败。该场景下，新的Transaction Coordinator会直接给 Producer 返回成功。

事务过期机制

事务超时

transaction.timeout.ms

终止过期事务

当 Producer 失败时，Transaction Coordinator必须能够主动的让某些进行中的事务过期。否则没有 Producer 的参与，Transaction Coordinator无法判断这些事务应该如何处理，这会造成：

如果这种进行中事务太多，会造成Transaction Coordinator需要维护大量的事务状态，大量占用内存
Transaction Log内也会存在大量数据，造成新的Transaction Coordinator启动缓慢
READ_COMMITTED的 Consumer 需要缓存大量的消息，造成不必要的内存浪费甚至是 OOM
如果多个Transaction ID不同的 Producer 交叉写同一个 Partition，当一个 Producer 的事务状态不更新时，READ_COMMITTED的 Consumer 为了保证顺序消费而被阻塞

为了避免上述问题，Transaction Coordinator会周期性遍历内存中的事务状态 Map，并执行如下操作

如果状态是BEGIN并且其最后更新时间与当前时间差大于transaction.remove.expired.transaction.cleanup.interval.ms（默认值为 1 小时），则主动将其终止：1）未避免原 Producer 临时恢复与当前终止流程冲突，增加该 Producer 对应的 PID 的 epoch，并确保将该更新的信息写入Transaction Log；2）以更新后的 epoch 回滚事务，从而使得该事务相关的所有 Broker 都更新其缓存的该 PID 的 epoch 从而拒绝旧 Producer 的写操作
如果状态是PREPARE_COMMIT，完成后续的 COMMIT 流程————向各<Topic, Partition>写入Transaction Marker，在Transaction Log内写入COMPLETE_COMMIT
如果状态是PREPARE_ABORT，完成后续 ABORT 流程

终止`Transaction ID`

某Transaction ID的 Producer 可能很长时间不再发送数据，Transaction Coordinator没必要再保存该Transaction ID与PID等的映射，否则可能会造成大量的资源浪费。因此需要有一个机制探测不再活跃的Transaction ID并将其信息删除。

Transaction Coordinator会周期性遍历内存中的Transaction ID与PID映射，如果某Transaction ID没有对应的正在进行中的事务并且它对应的最后一个事务的结束时间与当前时间差大于transactional.id.expiration.ms（默认值是 7 天），则将其从内存中删除并在Transaction Log中将其对应的日志的值设置为 null 从而使得 Log Compact 可将其记录删除。

PostgreSQL MVCC

Kafka 的事务机制与《 MVCC PostgreSQL 实现事务和多版本并发控制的精华》一文中介绍的 PostgreSQL 通过 MVCC 实现事务的机制非常类似，对于事务的回滚，并不需要删除已写入的数据，都是将写入数据的事务标记为 Rollback/Abort 从而在读数据时过滤该数据。

两阶段提交

Kafka 的事务机制与《分布式事务（一）两阶段提交及 JTA 》一文中所介绍的两阶段提交机制看似相似，都分 PREPARE 阶段和最终 COMMIT 阶段，但又有很大不同。

Kafka 事务机制中，PREPARE 时即要指明是PREPARE_COMMIT还是PREPARE_ABORT，并且只须在Transaction Log中标记即可，无须其它组件参与。而两阶段提交的 PREPARE 需要发送给所有的分布式事务参与方，并且事务参与方需要尽可能准备好，并根据准备情况返回Prepared或Non-Prepared状态给事务管理器。
Kafka 事务中，一但发起PREPARE_COMMIT或PREPARE_ABORT，则确定该事务最终的结果应该是被COMMIT或ABORT。而分布式事务中，PREPARE 后由各事务参与方返回状态，只有所有参与方均返回Prepared状态才会真正执行 COMMIT，否则执行 ROLLBACK
Kafka 事务机制中，某几个 Partition 在 COMMIT 或 ABORT 过程中变为不可用，只影响该 Partition 不影响其它 Partition。两阶段提交中，若唯一收到 COMMIT 命令参与者 Crash，其它事务参与方无法判断事务状态从而使得整个事务阻塞
Kafka 事务机制引入事务超时机制，有效避免了挂起的事务影响其它事务的问题
Kafka 事务机制中存在多个Transaction Coordinator实例，而分布式事务中只有一个事务管理器

Zookeeper

Zookeeper 的原子广播协议与两阶段提交以及 Kafka 事务机制有相似之处，但又有各自的特点

Kafka 事务可 COMMIT 也可 ABORT。而 Zookeeper 原子广播协议只有 COMMIT 没有 ABORT。当然，Zookeeper 不 COMMIT 某消息也即等效于 ABORT 该消息的更新。
Kafka 存在多个Transaction Coordinator实例，扩展性较好。而 Zookeeper 写操作只能在 Leader 节点进行，所以其写性能远低于读性能。
Kafka 事务是 COMMIT 还是 ABORT 完全取决于 Producer 即客户端。而 Zookeeper 原子广播协议中某条消息是否被 COMMIT 取决于是否有一大半 FOLLOWER ACK 该消息。

Kafka 系列文章

发布

暂无评论

2021 ThoughtWorks 技术雷达峰会

	Producer<String, String> producer = new KafkaProducer<String, String>(props);

	// 初始化事务，包括结束该 Transaction ID 对应的未完成的事务（如果有）
	// 保证新的事务在一个正确的状态下启动
	producer.initTransactions();

	// 开始事务
	producer.beginTransaction();

	// 消费数据
	ConsumerRecords<String, String> records = consumer.poll(100);

	try{
	// 发送数据
	producer.send(new ProducerRecord<String, String>("Topic", "Key", "Value"));

	// 发送消费数据的 Offset，将上述数据消费与数据发送纳入同一个 Transaction 内
	producer.sendOffsetsToTransaction(offsets, "group1");

	// 数据发送及 Offset 发送均成功的情况下，提交事务
	producer.commitTransaction();
	} catch (ProducerFencedException \| OutOfOrderSequenceException \| AuthorizationException e) {
	// 数据发送或者 Offset 发送出现异常时，终止事务
	producer.abortTransaction();
	} finally {
	// 关闭 Producer 和 Consumer
	producer.close();
	consumer.close();
	}

创作场景

Kafka 设计解析（八）：Kafka 事务机制与 Exactly Once 语义实现原理

Exactly Once

操作原子性

幂等性发送

事务性保证

事务性消息传递

事务中 Offset 的提交

用于事务特性的控制型消息

事务处理样例代码

完整事务过程

找到Transaction Coordinator

获取 PID

开启事务

Consume-Transform-Produce

Commit 或 Abort 事务

总结

Exception 处理

处理Transaction Coordinator失败

写PREPARE_COMMIT/PREPARE_ABORT前失败

写完PREPARE_COMMIT/PREPARE_ABORT后失败

写完COMPLETE_COMMIT/ABORT后失败

事务过期机制

事务超时

终止过期事务

终止Transaction ID

PostgreSQL MVCC

两阶段提交

Zookeeper

Kafka 系列文章

评论

更多内容推荐

推荐阅读

电子书

大厂实战PPT下载

`Exactly Once`

找到`Transaction Coordinator`

处理`Transaction Coordinator`失败

写`PREPARE_COMMIT/PREPARE_ABORT`前失败

写完`PREPARE_COMMIT/PREPARE_ABORT`后失败

写完`COMPLETE_COMMIT/ABORT`后失败

终止`Transaction ID`