Kafka 是如何实现事务的？

Kafka 是一个高度可扩展的分布式消息系统，在海量数据处理生态中占据着重要的地位。

数据处理的一个关键特性是数据的一致性。具体到 Kafka 的领域中，也就是生产者生产的数据和消费者消费的数据之间一对一的一致性。在各种类型的失败普遍存在的分布式系统环境下，保证业务层面一个整体的消息集合被原子的发布和恰好一次处理，是数据一致性在 Kafka 生态系统的实际要求。

本文介绍了 Kafka 生态中的事务机制的概念和流程。

Kafka 事务机制的概念

Kafka 从 0.11 版本开始支持了事务机制。Kafka 事务机制支持了跨分区的消息原子写功能。具体来说，Kafka 生产者在同一个事务内提交到多个分区的消息，要么同时成功，要么同时失败。这一保证在生产者运行时出现异常甚至宕机重启之后仍然成立。

此外，同一个事务内的消息将以生产者发送的顺序，唯一地提交到 Kafka 集群上。也就是说，事务机制从某种层面上保证了消息被恰好一次地提交到 Kafka 集群。众所周知，恰好一次送达在分布式系统中是不可能实现的。这个论断有一些微妙的名词重载问题，但大抵没错，所有声称能够做到恰好一次处理的系统都在某个地方依赖了幂等性。

Kafka 的事务机制被广泛用于现实世界中复杂业务需要保证一个业务领域中原子的概念被原子地提交的场景。

例如，一次下单流水包括订单生成消息和库存扣减消息，如果这两个消息在历史上由两个主题分管，那么它们在业务上的原子性就要求 Kafka 要利用事务机制原子地提交到 Kafka 集群上。

还有，对于复杂的流式处理系统，Kafka 生产者的上游可能是另一个流式处理系统，这个系统可能有着自己的一致性方案。为了跟上游系统的一致性方案协调，Kafka 就需要提供一个尽可能通用且易于组合的一致性机制，即灵活的事务机制，来帮助实现端到端的一致性。

Kafka 事务机制的流程

分布式系统的数据一致性是难的。要想理解一个系统提供何种程度的数据一致性保证，以及这样的保证对应用程序提出了什么样的要求，再及在哪些情况下一致性保证会出现什么方面的回退，细究其一致性机制的实现是必须的。

上面我们提到，事务机制的核心特征是能跨越多个分区原子地提交消息集合，甚至这些分区从属于不同的主题。同时，被提交的消息集合中的消息每条仅被提交一次，并保持它们在生产者应用中被生产的顺序写入到 Kafka 集群的消息日志中。此外，事务能够容忍生产者运行时出现异常甚至宕机重启。

实现事务机制最关键的概念就是事务的唯一标识符（ TransactionalID ），Kafka 使用 TransactionalID 来关联进行中的事务。TransactionalID 由用户提供，这是因为 Kafka 作为系统本身无法独立的识别出宕机前后的两个不同的进程其实是要同一个逻辑上的事务。

对于同一个生产者应用前后进行的多个事务，TransactionalID 并不需要每次都生成一个新的。这是因为 Kafka 还实现了 ProducerID 以及 epoch 机制。这个机制在事务机制中的用途主要是用于标识不同的会话，同一个会话 ProducerID 的值相同，但有可能有多个任期。ProducerID 仅在会话切换时改变，而任期会在每次新的事物初始化时被更新。这样，同一个 TransactionalID 就能作为跨会话的多个独立事务的标识。

接下来，我们从一个事务的完整流程出发讨论客户端也就是生产者和消费者，以及服务端也就是 Kafka 集群在这个流程中扮演了什么角色，执行了什么动作。

初始化事务上下文

逻辑上说，事务总是从生产者提起的。生产者通过调用 initTransactions 方法初始化事务上下文。首要做的事情就是找到 Kafka 集群负责管理当前事务的事务协调者（ TransactionCoordinator ），向其申请 ProducerID 资源。初始的 ProducerID 及 epoch 都是未初始化的状态。

生产者一侧的事务管理者（ TransactionManager ）收到相应的方法调用之后先后发送查找事务协调者的信息和初始化 ProducerID 的信息。事务相关的所有元数据信息都会由客户端即生产者一侧的事务管理者和服务端即 Kafka 集群的一个 Broker 上的事务协调者交互完成。

一开始，生产者并不知道哪个 Broker 上有自己 TransactionalID 关联的事务协调者。逻辑上，所有事务相关的需要持久化的数据最终都会写到一个特殊的主题 __transaction_state 上。这跟前面回答消费位点管理文章中的管理消费者消费位点的特殊主题 __consumer_offsets 构成了目前 Kafka 系统里唯二的特殊主题。

对于一个生产者或者说被 TransactionalID 唯一标识的事务来说，它的事务协调者就是该事务的元数据最终存储在 __transaction_state 主题上对应分区的分区首领。对于一个具体的事务来说，它的元数据将被其 TransactionalID 的哈希值的绝对值模分区数的分区所记录，这也是常见的确定分区的方案。

生产者将查找事务协调者的信息发送到集群的任意一个 Broker 上，由它计算出实际的事务协调者，获取对应的节点信息后返回给生产者。这样，生产者就找到了事务协调者。

随后，生产者会向事务协调者申请一个 ProducerID 资源，这个资源包括 ProducerID 和对应的 epoch 信息。事务协调者收到对应请求后，将会首先判断同一个 TransactionalID 下的事务的状态，以应对好跨会话的事务的管理。

第一步，事务协调者会获取 TransactionalID 对应的事务元数据信息。前面提到，这些元数据信息将被写在特殊主题 __transaction_state 上，这也是事务元数据信息对生产者和 Kafka 集群都容错的需要。

如果获取不到元数据信息，那么就初始化事务元数据信息，包括从获取一个新的 ProducerID 资源，并将它和 TransactionalID 以及分区编号和其他一些配置信息一起打包持久化。

其中，获取一个新的 ProducerID 资源需要 ProducerID 管理器从 ZooKeeper 上申请一个 ProducerID 的号段，在逐一的分配出去。申请号段的手段是修改 ZooKeeper 上 /latest_producer_id_block 节点的信息，流程是读节点上最后一个被申请的 ProducerID 的信息，加上要申请的号段的长度，再更新节点上最后一个被申请的 ProducerID 的信息。由于 ZooKeeper 对节点的更新有版本控制，因此并发的请求将会导致其中若干个请求目标版本失配，并提起重试。ProducerID 的长度是 Long 类型的长度，因此在实际使用过程中几乎不可能用完，Kafka 对号段资源耗尽的情况抛出致命错误并不尝试恢复。

如果获取到了相同 TransactionalID 先前的元数据信息，那么根据事务协调器事务先前的状态采取不同的行为。

1. 如果此时状态转移正在进行，直接返回 CONCURRENT_TRANSACTIONS 异常。 注意这里是事务协调器上正在 发生并发的状态转移。 通常来说，并发的状态转移应该依次执行，直接返回此异常可避免客户端即生产者请求超时，而是让生产者稍后自行重试。 这也是一种乐观的加 锁 策略。

2. 如果此时状态为 PrepareAbort 或 PrepareCommit 则返回 CONCURRENT_TRANSACTIONS 异常。 同 样的，此时状态即将转换为终结状态，无需强行终止先前的事务，否则将会产生无谓的浪费。

3. 如果此时状态为 Dead 或 PrepareEpochFence 或当前 ProducerID 和 epoch 对不上，直接抛出不可重试的异常。 这是由于要么是 先前的 Prod ucer 且已经被新的 Producer 替代，要么事务已经超时，无需再次尝试。

4. 如果此时状态为 Ongoing 则事务协调者会将事务转移到 PrepareEpochFence 状态，然后再丢弃当前的事务，并返回 CONCURRENT_TRANSACTIONS 异常。

5. 如果此时状态为 CompleteAbort 或 CompleteCommit 或 Empty 之一那么先将状态转移为 Empty 然后更新 epoch 值。

经过这么一连环的操作，Kafka 就将事务执行的上下文初始化好了。

开始一个事务

初始化事务的流程实际上是生产者和对应的事务协调者就事务状态达成一致，进入到一个可以提起新的事务的状态。此时，生产者可以通过 beginTransaction 方法开始一个事务操作。这个方法只会将本地事务状态转移到 IN_TRANSACTION 状态，在真正的提交事务中的消息之前，不会有跟 Kafka 集群的交互。

生产者将自己标记为开始事务之后，也就是本地事务状态转移到事务进行中的状态之后，就可以开始发送事务中的消息了。

发送事务中的消息

生产者在发送事务中的消息的时候，会将消息对应的分区添加到事务管理器中去，如果这个分区此前没被添加过，那么事务管理器会在下一次发送消息之前插入一条 AddPartitionsToTxnRequest 请求来告诉 Kafka 集群的事务协调者参与事务的分区的信息。事务协调者收到这条信息之后，将会更新事务的元数据，并将元数据持久化到 __transaction_state 中。

对于生产者发送的消息，仍然和一般的消息生产一样采用 ProduceRequest 请求。除了会在请求中带上相应的 TransactionalID 信息和属于事务中的消息的标识符，它跟生产者生产的普通信息别无二致。如果消费者没有配置读已提交的隔离级别，那么这些消息在被 Kafka 集群接受并持久化到主题分区中时，就已经对消费者可见而且可以被消费了。

事务中的消息的顺序性保证也是在发送事务的时候检查的。

生产者此时已经申请到了一个 ProducerID 资源，当它向一个分区发送消息时，内部会有一个消息管理器为每个不同的分区维护一个顺序编号（ SequenceNumber ）。相应地，Kafka 集群也会为每个 ProducerID 到每个分区的消息生产维护一个顺序编号。

ProducerRequest 请求中包含了顺序编号信息。如果 Kafka 集群看到请求的顺序编号跟自己的顺序编号是连续的，即比自己的顺序编号恰好大一，那么接受这条消息。否则，如果请求的顺序编号大一以上，则说明是一个乱序的消息，直接拒绝并抛出异常。如果请求的顺序编号相同或更小，则说明是一个重复发送的消息，直接忽略并告诉客户端是一个重复消息。

提交事务

在一个事务相关的所有消息都发送完毕之后，生产者就可以调用 commitTransaction 方法来提交整个事务了。对于事务中途发生异常的情形，也可以通过调用 abortTransaction 来丢弃整个事务。这两个操作都是将事务状态转移到终结状态，彼此之间有许多相似点。

无论是提交还是丢弃，生产者都是给事务协调者发送 EndTxnRequest 请求，请求中包含一个字段来判断是提交还是丢弃。事务协调者在收到这个请求后，首先更新事务状态到 PrepareAbort 或 PrepareCommit 并更新状态到 __transaction_state 中。

如果在状态更新成功前事务协调者宕机，那么恢复过来的事务协调者将认为事务在 Ongoing 状态中，此时生产者由于收不到确认回复，会重试 EndTxnRequest 请求，并最终更新事务到 PrepareAbort 或 PrepareCommit 状态。

随后，根据是提交还是丢弃，分别向事务涉及到的所有分区的分区首领发送事务标志（ TransactionMarker ）。

事务标志是 Kafka 事务机制引入的不同于业务消息的事务控制消息。它的作用主要是标识事务已经完成，这个消息同业务消息一样能够被消费者所消费，并且它和事务中的业务消息能够通过 TransactionalID 关联起来，从而支持配置了读已提交特性的消费者忽略尚未提交的事务消息或被丢弃的事务消息。

如果在事务标志写到涉及到的所有分区的分区首领之前，事务协调者宕机或者分区首领宕机或网络分区，新起来的事务协调者或超时后重试的事务协调者会重新向分区首领写入事务标志。事务标志是幂等的，因此不会影响事务提交的结果。这里我们印证了之前所说的所有声称能够做到恰好一次处理的系统都在某个地方依赖了幂等性。

在当前事务涉及到的所有分区都已经把事务标志信息持久化到主题分区之后，事务协调者才会将这个事务的状态置为提交或丢弃，并持久化到事务日志文件中。在这之后，一个 Kafka 事务才算真正的完成了。事务协调者中缓存的关于当前事务的元数据就可以清理了。

如果在事务协调者回复生产者提交成功之前宕机，在恢复之后生产者再次提交事务时会直接返回事务提交成功。

总的来说，事务的状态以 __transaction_state 主题上持久化的元数据信息为准。

超时过期事务

分布式系统由于天然的网络阻塞或分区等失败原因，操作在成功和失败之外还有超时这第三种状态。现实中的分布式系统必须合理地处理超时的状态，否则永久阻塞或等待在任何实际的业务领域中都是不可接受的。

Kafka 事务机制本身可以配置事务超时，在事务管理者和事务协调者交互的各个过程中都会检验事务超时的配置，如果事务已经超时则抛出异常。

但是，在网络分区的情况下，可能 Kafka 集群根本就等不到生产者发送的消息。这个时候，Kafka 集群就需要相应的机制来主动过期。否则永不过期的中间状态事务在生产者宕机且不可恢复或不再恢复的情况下将逐步积累成存储垃圾。

Kafka 集群会周期性的轮询内存中的事务信息。如果发现进行中的事务最后的状态更新时间距今已经超过了配置的集群事务清理时间阈值，则采取丢弃该事务的操作。同时，为了避免操作过程中并发地收到原 Producer 发来事务更新请求，首先更新事务关联的 ProducerID 的 epoch 以将原 Producer 的 epoch 隔离掉。换个角度说，也就是以一个新的有效的身份执行丢弃事务操作，以免分不清到底是谁在丢弃事务。

此外，轮询中还会检查 TransactionalID 最新的事务信息，如果一个 TransactionalID 最后一个事务距今已经已经超过了配置的集群 TransactionalID 清理时间阈值，则将该 TransactionalID 对应的元数据信息都进行清理。

上面的讨论中还有两个重要的主题被忽略了。一个是 Kafka 事务机制支持在同一个事务里进行消息生产和消息消费位点提交，另一个是配置了读已提交的消费者如何在事务未提交以及丢弃事务时正确的读取事务中消息。

前者不是特别复杂，只需要将消费位点提交视作一条事务中的消息，和消息生产以及控制消息同等待遇，在提交的时候也被事务标志所界定即可。

不展开聊是因为这个特性通常只在仅适用 Kafka 搭建流式处理流水线的场景下有用，尤其是 Kafka Streams 解决方案。

对于组合多个系统的流式处理流水线来说，消息从 Kafka 中消费得到是上游，生产到 Kafka 上是下游，中间是另一个例如 Flink 的流式计算系统。在这种场景下，消费位点的管理和事务地生产消息是两个可以分开考虑的事情，可以跟其他系统的一致性方案例如 Flink 的 Checkpoint 机制相结合，而不需要非得在同一个事务里既提交消费位点，又提交新的消息。

后者主要靠 Kafka 集群在管理消费位点拉取请求的时候，通过随事务机制的引入新添加的 LastStableOffset 概念来响应配置为读已提交的消费者的请求。在事务完成之前不会允许读已提交的消费者拉取事务中的消息。显然，这有可能导致消费者拉取新消息时长时间的阻塞。因此在实践中应当尽量避免长时间的事务。

对于丢弃事务的消息，Kafka 集群会维护一个丢弃事务的消息的元数据，从而支持消费者同时拉取消息和丢弃事务的消息的元数据，自行比对筛掉丢弃事务的消息。在正常的业务场景里，丢弃的事务不会太多，从而维护这样的一份元数据以及让消费者自行筛选会是一个能够接受的选择。