Kafka Exactly-Once 之事务性实现

kafka事务解决什么问题？

1. 在写多个 Topic-Partition 时，执行的一批写入操作，有可能出现部分 Topic-Partition 写入成功，部分写入失败（比如达到重试次数），这相当于出现了中间的状态，这并不是我们期望的结果；

2. Producer 应用中间挂之后再恢复，无法做到 Exactly-Once 语义保证；

由于幂等性无法提供多会话，以及多分区的Exactly-Once语义，所以事务性就是用来解决这个问题。因此，我们是需要一种更强的事务保证，能够原子处理多个 Partition 的写入操作，数据要么全部写入成功，要么全部失败，不期望出现中间状态。这就是 Kafka Transactions 希望解决的问题。但是事务性的实现要比幂等性的实现复杂很多。

本文以 Apache Kafka 2.0.0 代码实现为例，深入分析一下 Kafka 是如何实现这一机制的。

Kafka Transactions

kafka事务处理主要是消费和生产的batch处理（涉及多个partition）在一个原子操作内完成，操作要么全部完成、要么全部失败。wield实现这几机制我们需要提供一个唯一id 并且这个id在故障恢复后也不能改变，这个 id 就是 TransactionnalId（也叫 txn.id，后面会详细讲述），txn.id 可以跟内部的 PID 1:1 分配，它们不同的是 txn.id 是用户提供的，而 PID 是 Producer 内部自动生成的（并且故障恢复后这个 PID 会变化），有了 txn.id 这个机制，就可以实现多 partition、跨会话的 EOS 语义。

当用户使用 Kafka 的事务性时，Kafka 可以做到的保证：

1. 跨会话的幂等性写入：即使中间故障也能保证上一次幂等性的实现。

2. 跨会话的幂等性事务恢复：即使上一个应用实例挂了，启动的下一个实例依然可以保证上一个事务完成（commit 或者 abort）；

3. 跨多个 Topic-Partition 的幂等性写入，Kafka 可以保证跨多个 Topic-Partition 的数据要么全部写入成功，要么全部失败，不会出现中间状态。

上面是从 Producer 的角度来看，那么如果从 Consumer 角度呢？Consumer 端很难保证一个已经 commit 的事务的所有 msg 都会被消费，有以下几个原因：

1. 一个事务内的数据，可能会跨多个 log segment，如果旧的 segmeng 数据由于过期而被清除，那么这个事务的一部分数据就无法被消费到了；

2. Consumer 在消费时可以通过 seek 机制，随机从一个位置开始消费，这也会导致一个事务内的部分数据无法消费；

3. Consumer 可能没有订阅这个事务涉及的全部 Partition。\

事务性示例

Kafka 事务性的使用方法也非常简单，用户只需要在 Producer 的配置中配置 transactional.id，通过 initTransactions() 初始化事务状态信息，再通过 beginTransaction() 标识一个事务的开始，然后通过 commitTransaction() 或 abortTransaction() 对事务进行 commit 或 abort，示例如下所示：

Properties props = new Properties();props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");props.put("client.id", "ProducerTranscationnalExample");props.put("bootstrap.servers", "localhost:9092");props.put("transactional.id", "test-transactional");props.put("acks", "all");KafkaProducer producer = new KafkaProducer(props);producer.initTransactions();try { String msg = "matt test"; producer.beginTransaction(); producer.send(new ProducerRecord(topic, "0", msg.toString())); producer.send(new ProducerRecord(topic, "1", msg.toString())); producer.send(new ProducerRecord(topic, "2", msg.toString())); producer.commitTransaction();} catch (ProducerFencedException e1) { e1.printStackTrace(); producer.close();} catch (KafkaException e2) { e2.printStackTrace(); producer.abortTransaction();}producer.close();

事务性实现的关键

对于 Kafka 的事务性实现，最关键的就是其事务操作原子性的实现。对于一个事务操作而言，其会涉及到多个 Topic-Partition 数据的写入，如果是一个 long transaction 操作，可能会涉及到非常多的数据，如何才能保证这个事务操作的原子性（要么全部完成，要么全部失败）呢？

1. 关于这点，最容易想到的应该是引用 2PC 协议（它主要是解决分布式系统数据一致性的问题）中协调者的角色，它的作用是统计所有参与者的投票结果，如果大家一致认为可以 commit，那么就执行 commit，否则执行 abort：

   a 我们来想一下，Kafka 是不是也可以引入一个类似的角色来管理事务的状态，只有当 Producer 真正 commit 时，事务才会提交，否则事务会还在进行中（实际的实现中还需要考虑 timeout 的情况），不会处于完成状态；

   b Producer 在开始一个事务时，告诉【协调者】事务开始，然后开始向多个 Topic-Partition 写数据，只有这批数据全部写完（中间没有出现异常），Producer 会调用 commit 接口进行 commit，然后事务真正提交，否则如果中间出现异常，那么事务将会被 abort（Producer 通过 abort 接口告诉【协调者】执行 abort 操作）；

    c 这里的协调者与 2PC 中的协调者略有不同，主要为了管理事务相关的状态信息，这就是 Kafka Server 端的 TransactionCoordinator 角色；

2. 有了上面的机制，是不是就可以了？很容易想到的问题就是 TransactionCoordinator 挂的话怎么办？TransactionCoordinator 如何实现高可用？

   a TransactionCoordinator 需要管理事务的状态信息，如果一个事务的 TransactionCoordinator 挂的话，需要转移到其他的机器上，这里关键是在 事务状态信息如何恢复？ 也就是事务的状态信息需要很强的容错性、一致性；

   b 关于数据的强容错性、一致性，存储的容错性方案基本就是多副本机制，而对于一致性，就有很多的机制实现，其实这个在 Kafka 内部已经实现（不考虑数据重复问题），那就是 min.isr + ack 机制；

   c 分析到这里，对于 Kafka 熟悉的同学应该就知道，这个是不是跟 __consumer_offset 这个内部的 topic 很像，TransactionCoordinator 也跟 GroupCoordinator 类似，而对应事务数据（transaction log）就是 __transaction_state 这个内部 topic，所有事务状态信息都会持久化到这个 topic，TransactionCoordinator 在做故障恢复也是从这个 topic 中恢复数据；

3. 有了上面的机制，就够了么？我们再来考虑一种情况，我们期望一个 Producer 在 Fail 恢复后能主动 abort 上次未完成的事务（接上之前未完成的事务），然后重新开始一个事务，这种情况应该怎么办？之前幂等性引入的 PID 是无法解决这个问题的，因为每次 Producer 在重启时，PID 都会更新为一个新值：

   Kafka 在 Producer 端引入了一个 TransactionalId 来解决这个问题，这个 txn.id 是由应用来配置的；

4. 再来考虑一个问题，在具体的实现时，我们应该如何标识一个事务操作的开始、进行、完成的状态？正常来说，一个事务操作是由很多操作组成的一个操作单元，对于 TransactionCoordinator 而言，是需要准确知道当前的事务操作处于哪个阶段，这样在容错恢复时，新选举的 TransactionCoordinator 才能恢复之前的状态：

    这个就是事务状态转移，一个事务从开始，都会有一个相应的状态标识，直到事务完成，有了事务的状态转移关系之后，TransactionCoordinator 对于事务的管理就会简单很多，TransactionCoordinator 会将当前事务的状态信息都会缓存起来，每当事务需要进行转移，就更新缓存中事务的状态（前提是这个状态转移是有效的）。

下事务性实现的一些关键的实现机制

TransactionCoordinator

TransactionCoordinator 与 GroupCoordinator 有一些相似之处，它主要是处理来自 Transactional Producer 的一些与事务相关的请求，

总结一下，TransactionCoordinator 主要的功能有三个，分别是：

1. 处理事务相关的请求；

2. 维护事务的状态信息；

3. 向其他 Broker 发送 Transaction Marker 数据。

Transaction Log（__transaction_state）

在前面分析中，讨论过一个问题，那就是如果 TransactionCoordinator 故障的话应该怎么恢复？

正如 GroupCoordinator 的实现一样，TransactionCoordinator 的恢复也是通过 __transaction_state中读取之前事务的日志信息，来恢复其状态信息，前提是要求事务日志写入做相应的不丢配置。这也是 __transaction_state 一个重要作用之一，用于 TransactionCoordinator 的恢复，__transaction_state 与 __consumer_offsets 一样是 compact 类型的 topic

怎么恢复之前的状态？我们知道 Kafka 内部有一个事务 topic __transaction_state，一个事务应该由哪个 TransactionCoordinator 来处理，是根据其 txn.id 的 hash 值与 __transaction_state 的 partition 数取模得到，__transaction_state Partition 默认是50个，假设取模之后的结果是2，那么这个 txn.id 应该由 __transaction_state Partition 2 的 leader 来处理。

transaction_state 这个 topic 默认是由 Server 端的 transaction.state.log.replication.factor 参数来配置，默认是3

Transaction Marker

Transaction Marker 也叫做 control messages，它的作用主要是告诉这个事务操作涉及的 Topic-Partition Set 的 leaders 当前的事务操作已经完成，可以执行 commit 或者 abort（Marker 主要的内容就是 commit 或 abort），这个 marker 数据由该事务的 TransactionCoordinator 来发送的。我们来假设一下：如果没有 Transaction Marker，一个事务在完成后，如何执行 commit 操作？（以这个事务涉及多个 Topic-Partition 写入为例）

1. Transactional Producer 在进行 commit 时，需要先告诉 TransactionCoordinator 这个事务可以 commit 了（因为 TransactionCoordinator 记录这个事务对应的状态信息），然后再去告诉这些 Topic-Partition 的 leader 当前已经可以 commit，也就是 Transactional Producer 在执行 commit 时，至少需要做两步操作；

2. 在 Transactional Producer 通知这些 Topic-Partition 的 leader 事务可以 commit 时，这些 Topic-Partition 应该怎么处理呢？难道是 commit 时再把数据持久化到磁盘，abort 时就直接丢弃不做持久化？这明显是问题的，如果这是一个 long transaction 操作，写数据非常多，内存中无法存下，数据肯定是需要持久化到硬盘的，如果数据已经持久化到硬盘了，假设这个时候收到了一个 abort 操作，是需要把数据再从硬盘清掉？

        这种方案有一个问题是：已经持久化的数据是持久化到本身的日志文件，还是其他文件？如果持久化本来的日志文件中，那么 consumer 消费到一个未 commit 的数据怎么办？这些数据是有可能 abort 的，如果是持久化到其他文件中，这会涉及到数据多次写磁盘、从磁盘清除的操作，会影响其 server 端的性能；

再看下如果有了 Transaction Marker 这个机制后，情况会变成什么样？

1. 首先 Transactional Producer 只需要告诉 TransactionCoordinator 当前事务可以 commit，然后再由 TransactionCoordinator 来向其涉及到的 Topic-Partition 的 leader 发送 Transaction Marker 数据，这里减轻了 Client 的压力，而且 TransactionCoordinator 会做一些优化，如果这个目标 Broker 涉及到多个事务操作，是可以共享这个 TCP 连接的；

2. 有了 Transaction Marker 之后，Producer 在持久化数据时就简单很多，写入的数据跟之前一样，按照条件持久化到硬盘（数据会有一个标识，标识这条或这批数据是不是事务写入的数据），当收到 Transaction Marker 时，把这个 Transaction Marker 数据也直接写入这个 Partition 中，这样在处理 Consumer 消费时，就可以根据 marker 信息做相应的处理。

Empty  Ongoing PrepareCommit PrepareAbort CompleteCommit CompleteAbort Dead PrepareEpochFence

Client 端事务状态管理

Client 的事务状态信息主要记录本地事务的状态，当然跟其他的系统类似，本地的状态信息与 Server 端的状态信息并不完全一致（状态的设置，就像 GroupCoodinator 会维护一个 Group 的状态，每个 Consumer 也会维护本地的 Consumer 对象的状态一样）。Client 端的事务状态信息主要用于 Client 端的事务状态处理，其主要有以下几种：

事务性的整体流程

1. Finding a TransactionCoordinator

对于事务性的处理，第一步首先需要做的就是找到这个事务 txn.id 对应的 TransactionCoordinator，Transaction Producer 会向 Broker （随机选择一台 broker，一般选择本地连接最少的这台 broker）发送 FindCoordinatorRequest 请求，获取其 TransactionCoordinator。

怎么找到对应的 TransactionCoordinator 呢？这个前面已经讲过了，主要是通过下面的方法获取 __transaction_state 的 Partition，该 Partition 对应的 leader 就是这个 txn.id 对应的 TransactionCoordinator。

2. Getting a PID

Transaction Producer 在 initializeTransactions() 方法中会向 TransactionCoordinator 发送 InitPidRequest 请求获取其分配的 PID，

   1. 如果这个 txn.id 之前没有相应的事务状态（new txn.id），那么会初始化其事务 meta 信息 TransactionMetadata（会给其分配一个 PID，初始的 epoch 为-1），如果有事务状态，获取之前的状态；

   2. 校验其 TransactionMetadata 的状态信息

   3. 将 txn.id 与相应的 TransactionMetadata 持久化到事务日志中，对于 new txn.id，这个持久化的数据主要时 txn.id 与 pid 关系信息

3. Starting a Transaction

前面两步都是 Transaction Producer 调用 initTransactions() 部分，到这里，Producer 可以调用 beginTransaction() 开始一个事务操作，其实现方法如下面所示：

//KafkaProducer//note: 应该在一个事务操作之前进行调用public void beginTransaction() throws ProducerFencedException { throwIfNoTransactionManager(); transactionManager.beginTransaction();}// TransactionManager//note: 在一个事务开始之前进行调用，这里实际上只是转换了状态（只在 producer 本地记录了状态的开始）public synchronized void beginTransaction() { ensureTransactional(); maybeFailWithError(); transitionTo(State.IN_TRANSACTION);}

4. Consume-Porcess-Produce Loop

在这个阶段，Transaction Producer 会做相应的处理，主要包括：从 consumer 拉取数据、对数据做相应的处理、通过 Producer 写入到下游系统中

kafka事务流程

Kafka消息事务

01基本概念

为了支持事务，Kafka 0.11.0版本引入以下概念：

1.事务协调者：类似于消费组负载均衡的协调者，每一个实现事务的生产端都被分配到一个事务协调者(Transaction Coordinator)。

2.引入一个内部Kafka Topic作为事务Log：类似于消费管理Offset的Topic，事务Topic本身也是持久化的，日志信息记录事务状态信息，由事务协调者写入。

3.引入控制消息(Control Messages)：这些消息是客户端产生的并写入到主题的特殊消息，但对于使用者来说不可见。它们是用来让broker告知消费者之前拉取的消息是否被原子性提交。

4.引入TransactionId：不同生产实例使用同一个TransactionId表示是同一个事务，可以跨Session的数据幂等发送。当具有相同Transaction ID的新的Producer实例被创建且工作时，旧的且拥有相同Transaction ID的Producer将不再工作，避免事务僵死。

5.Producer ID：每个新的Producer在初始化的时候会被分配一个唯一的PID，这个PID对用户是不可见的。主要是为提供幂等性时引入的。

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

7.每个生产者增加一个epoch：用于标识同一个事务Id在一次事务中的epoch，每次初始化事务时会递增，从而让服务端可以知道生产者请求是否旧的请求。

8.幂等性：保证发送单个分区的消息只会发送一次，不会出现重复消息。增加一个幂等性的开关enable.idempotence，可以独立与事务使用，即可以只开启幂等但不开启事务。

如下图所示：

1、查找事务协调者

生产者会首先发起一个查找事务协调者的请求(FindCoordinatorRequest)。协调者会负责分配一个PID给生产者。类似于消费组的协调者。

2、获取produce ID

在知道事务协调者后，生产者需要往协调者发送初始化pid请求(initPidRequest)。这个请求分两种情况：

●不带transactionID

这种情况下直接生成一个新的produce ID即可，返回给客户端

●带transactionID

这种情况下，kafka根据transactionalId获取对应的PID，这个对应关系是保存在事务日志中（上图2a）。这样可以确保相同的TransactionId返回相同的PID，用于恢复或者终止之前未完成的事务。

3、启动事务

生产者通过调用beginTransaction接口启动事务，此时只是内部的状态记录为事务开始，但是事务协调者认为事务开始只有当生产者开始发送第一条消息才开始。

4、消费和生产配合过程

这一步是消费和生成互相配合完成事务的过程，其中涉及多个请求：

●增加分区到事务请求

当生产者有新分区要写入数据，则会发送AddPartitionToTxnRequest到事务协调者。协调者会处理请求，主要做的事情是更新事务元数据信息，并把信息写入到事务日志中（事务Topic）。

●生产请求

生产者通过调用send接口发送数据到分区，这些请求新增pid，epoch和sequence number字段。

●增加消费offset到事务

生产者通过新增的snedOffsets ToTransaction接口，会发送某个分区的Offset信息到事务协调者。协调者会把分区信息增加到事务中。

●事务提交offset请求

当生产者调用事务提交offset接口后，会发送一个TxnOffsetCommitRequest请求到消费组协调者，消费组协调者会把offset存储在__consumer-offsets Topic中。协调者会根据请求的PID和epoch验证生产者是否允许发起这个请求。 消费offset只有当事务提交后才对外可见。

 5、提交或回滚事务

用户通过调用commitTransaction或abortTranssaction方法提交或回滚事务。

●EndTxnRequest

当生产者完成事务后，客户端需要显式调用结束事务或者回滚事务。前者会使得消息对消费者可见，后者会对生产数据标记为Abort状态，使得消息对消费者不可见。无论是提交或者回滚，都是发送一个EndTnxRequest请求到事务协调者，写入PREPARE_COMMIT或者PREPARE_ABORT信息到事务记录日志中(5.1a)。

●WriteTxnMarkerRequest

这个请求是事务协调者向事务中每个TopicPartition的Leader发送的。每个Broker收到请求后会写入COMMIT(PID)或者ABORT(PID)控制信息到数据日志中(5.2a)。

这个信息用于告知消费者当前消息是哪个事务，消息是否应该接受或者丢弃。而对于未提交消息，消费者会缓存该事务的消息直到提交或者回滚。

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

●写入最终提交或回滚信息

当提交和回滚信息写入数据日子后，事务协调者会往事务日志中写入最终的提交或者终止信息以表示事务已经完成(图5.3)，此时大部分于事务有关系的消息都可以被删除（通过标记后面在日志压缩时会被移除），我们只需要保留事务ID以及其时间戳即可。