5.Kafka客户端事务原理

今天药忘吃喽~ 2022-10-09 13:29 195阅读 0赞

### 文章目录 ###

*  kafka 客户端事务实现原理
 *   *  Producer顶层实现
     *  事务流程
     *   *  1.选取TC服务地址
         *  2.事务初始化
         *  3.发送消息给Topic
         *  4.发送提交请求
         *  5.持久化Commit请求
         *  6.发送事务结果信息给分区
     *  事务状态转换
     *   *  事务初始化请求
         *  开始事务请求
         *  事务分区上传请求
         *  事务消费位置添加请求
         *  事务消费位置提交请求
         *  事务提交或回滚请求
     *  整体流程图

# kafka 客户端事务实现原理 #

`kafka事务在流处理中应用很广泛，比如原子性的读取消息，立即发送消息，如果中途出现错误，支持会滚操作。`这里会讲解一下Kafka事务是如何实现的。

## Producer顶层实现 ##

事务的客户端中，只能是Producer（KafkaProducer的顶层实现）

public interface Producer<K, V> extends Closeable { 
        //事务初始化，包括申请 producer id
        void initTransactions();
        
        //开启事务，这里会更改事务的本地状态
        void beginTransaction() throws ProducerFencedException;
        
        //提交offset，offsets表示每个分区的消费位置， consumerGroupId表示消费组的名称
        void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets, String consumerGroupId) throws ProducerFencedException;
        //同上
        void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets, ConsumerGroupMetadata groupMetadata) throws ProducerFencedException;
    
        //事务提交（提交）
        void commitTransaction() throws ProducerFencedException;
        //放弃事务（回滚）
        void abortTransaction() throws ProducerFencedException;
    }

KafkaProducer类实现了Producer接口，实现基本上都比较简单，都是调用的transactionManager方法（事务处理代码都是基于TransactionCoordinator类里，后续会讲）

public void initTransactions() { 
            throwIfNoTransactionManager();
            throwIfProducerClosed();
            TransactionalRequestResult result = transactionManager.initializeTransactions();
            sender.wakeup();
           	result.await(maxBlockTimeMs, TimeUnit.MILLISECONDS);
    		}
    
    public void beginTransaction() throws ProducerFencedException { 
            throwIfNoTransactionManager();
            throwIfProducerClosed();
            transactionManager.beginTransaction();
        }
    
    public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,
                                             ConsumerGroupMetadata groupMetadata) throws ProducerFencedException { 
            throwIfInvalidGroupMetadata(groupMetadata);
            throwIfNoTransactionManager();
            throwIfProducerClosed();
            TransactionalRequestResult result = transactionManager.sendOffsetsToTransaction(offsets, groupMetadata);
            sender.wakeup();
            result.await(maxBlockTimeMs, TimeUnit.MILLISECONDS);
        }
    
    public void commitTransaction() throws ProducerFencedException { 
            throwIfNoTransactionManager();
            throwIfProducerClosed();
            TransactionalRequestResult result = transactionManager.beginCommit();
            sender.wakeup();
            result.await(maxBlockTimeMs, TimeUnit.MILLISECONDS);
        }
    
    public void abortTransaction() throws ProducerFencedException { 
            throwIfNoTransactionManager();
            throwIfProducerClosed();
            log.info("Aborting incomplete transaction");
            TransactionalRequestResult result = transactionManager.beginAbort();
            sender.wakeup();
            result.await(maxBlockTimeMs, TimeUnit.MILLISECONDS);
        }

## 事务流程 ##

为了让更好的理解，我们先梳理一下客户端事务的状态流转

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxNDMwNjY1_size_16_color_FFFFFF_t_70_pic_center]

上图中的TC（**Transaction Coordinator** ）运行在kafka的服务端，\_\_transaction\_status 是TC服务持久化事务信息的topic名称

**当我们看到整体流程的时候是不是有些熟悉了，这就是最典型的2PC请求逻辑。**

### 1.选取TC服务地址 ###

**Producer会首先从Kafka集群中选择任意一台机器，然后发送选取TC服务的地址。kafka中有个特殊的topic:\_\_transaction\_status，负责持久化事务消息。这个topic跟 \_\_consumer\_offsets类似，默认有50个分区，每个分区负责一部分事务。事务划分根据transaction id，计算出该事务属于哪个分区，这个分区leader所在的机器就是负责这个事务的TC。**

### 2.事务初始化 ###

Producer在使用事务功能，必须先自定义一个**唯一的transaction id**，有了transaction id，即使客户端挂了，它重启后也能继续处理未完成的事务。

Kakfa实现事务需要依靠幂等性，而幂等性需要指定producer id（**因为0.11版本以后增加了幂等性传递选项，broker给每个producer都分配了ID，并且producer给每条被发送的消息都分配了一个序列号来避免产生重复消息，所以只有指定了**），所以在启动事务前，需要向TC申请producer id（图上1的流程）。TC在分配producer id后，会将它持久化到事务topic。

### 3.发送消息给Topic ###

Producer在接受到producer id 后就可以正常的发送消息了，不过发送消息前，需要先将这些消息的分区地址发送给TC（图上2.1的流程），TC服务会将分区地址持久化到\_\_transaction\_status，然后producer才能正常开始发送消息。**这些消息与普通消息不太相同，他们都会有一个表明自身是事务消息的字段**

那如何原子性的从某个topic读取消息，然后在发送到另外一个topic呢？

提交消费位置请求，可以用于原子性的从某个topic 读取消息，并且发送到另外一个topic，**我们知道一般的消费者使用消费组订阅topic，才会发送提交消费位置请求**，而这里由producer发送的。Producer首先回发送一条消息，里面会包含这个消费组对应的分区（每个消息组的消费位置都保存在\_\_consumer\_offset topic 的一个分区里），TC服务会将分区持久化之后，发送响应。Producer收到响应后，就会直接发送消费位置给GC(Group Coordinator)

### 4.发送提交请求 ###

Producer 发送完消息后，如果认为该事务可以提交了，就会发送提交请求给TC，TC只需要等待响应。Producer 在发送事务提交请求之前，会等待之前所有的请求都已经发送并且响应成功。（**这里是不是就是2PC的逻辑了**）

### 5.持久化Commit请求 ###

TC服务收到事务提交请求后，会将提交信息先持久化到\*\*\_\_transaction\_status topic\*\*,持久化成功后，服务端就立即发送响应给Producer，然后找到该事务涉及到的分区，为每个分区生成提交请求，存到队列里等待发送。（**收到所有参与者响应后，持久化事务为提交成功状态。**）

**之前我们讲分布式事务篇章的时候，如果事务协调者在发送提交请求前，挂掉是不是的时候是不是可以考虑引入事务请求日志来处理该问题？**

Kafka就是这样保证事务协调者请求，协调者挂了后，新的协调者会读取\_\_transaction\_status topic的事务信息，如果只有事务提交信息，而没有事务完成消息，说明存在事务结果信息没有提交给分区，再进行处理就可以了。

### 6.发送事务结果信息给分区 ###

后代线程会不停的从队列中，拉取请求并且发送到分区，当一个分区收到事务结果消息后，会将结果保存到分区里，并且返回成功响应到TC服务。当TC服务收到所有分区的成功响应后，会持久化一条事务完成的消息到\_\_transaction\_status topic，一个完整事务就完成了。

## 事务状态转换 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxNDMwNjY1_size_16_color_FFFFFF_t_70_pic_center 1]

在讲代码之前我们先了解一下client内的事务请求类

![在这里插入图片描述][20210625013434551.png_pic_center]

<table> 
 <thead> 
  <tr> 
   <th>请求类型</th> 
   <th>请求类</th> 
   <th>响应处理类</th> 
   <th>优先级（小则高）</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>事务寻找TC服务地址</td> 
   <td>FindCoordinatorRequest</td> 
   <td>FindCoordinatorHandler</td> 
   <td>0</td> 
  </tr> 
  <tr> 
   <td>事务初始化请求</td> 
   <td>InitProducerIdRequest</td> 
   <td>InitProducerIdHandler</td> 
   <td>1/3（如果没有初始化ProducerId则为1级）</td> 
  </tr> 
  <tr> 
   <td>事务消费位置提交请求</td> 
   <td>TxnOffsetCommitRequest</td> 
   <td>TxnOffsetCommitHandler</td> 
   <td>2</td> 
  </tr> 
  <tr> 
   <td>事务消费位置添加请求</td> 
   <td>AddOffsetsToTxnRequest</td> 
   <td>AddOffsetsToTxnHandler</td> 
   <td>2</td> 
  </tr> 
  <tr> 
   <td>事务分区上传请求</td> 
   <td>AddPartitionsToTxnRequest</td> 
   <td>AddPartitionsToTxnHandler</td> 
   <td>2</td> 
  </tr> 
  <tr> 
   <td>事务提交或回滚请求</td> 
   <td>EndTxnRequest</td> 
   <td>EndTxnHandler</td> 
   <td>3</td> 
  </tr> 
 </tbody> 
</table>

顶层类：TxnRequestHandler

abstract class TxnRequestHandler implements RequestCompletionHandler { 
            protected final TransactionalRequestResult result;
            private boolean isRetry = false;
            //...省略各种方法
      			//当请求完成后，回调处理逻辑，这里利用回调来调用handleResponse 方法
            @Override
            public void onComplete(ClientResponse response) { 
                if (response.requestHeader().correlationId() != inFlightRequestCorrelationId) { 
                    fatalError(new RuntimeException("Detected more than one in-flight transactional request."));
                } else { 
                  //...省略各种判断
                    if (response.hasResponse()) { 
                        requestBuilder());
                        synchronized (TransactionManager.this) { 
                            handleResponse(response.responseBody());
                        }
                    } 
                }
            }
    				// coordinatorType为GC（Group Coordinator ）与TC（Group Transcation）
            boolean needsCoordinator() { 
                return coordinatorType() != null;
            }
      			// 顶层请求建造者（每个请求的结构图不相同）
            abstract AbstractRequest.Builder<?> requestBuilder();
    				// 处理broker返回的结果
            abstract void handleResponse(AbstractResponse responseBody);
    				// 当前请求处理器（RequestHandler）的优先级
            abstract Priority priority();
        }

### 事务初始化请求 ###

InitProducerIdHandler：主要负责事务初始化，它会发送InitProducerIdRequest请求，向服务器获取到product id 以及epoch

synchronized TransactionalRequestResult initializeTransactions(ProducerIdAndEpoch producerIdAndEpoch) { 
            boolean isEpochBump = producerIdAndEpoch != ProducerIdAndEpoch.NONE;
            // handleCachedTransactionRequestResult方法 会判断pendingResult不为空且当前事务状态为 INITIALIZING 状态 则直接返回状态
            // 否则 构建 初始化ProductId的请求等方法。
            return handleCachedTransactionRequestResult(() -> { 
                // If this is an epoch bump, we will transition the state as part of handling the EndTxnRequest
                // isEpochBump 是否有获取过producerId
                if (!isEpochBump) { 
                    transitionTo(State.INITIALIZING);
                    log.info("Invoking InitProducerId for the first time in order to acquire a producer ID");
                } 
                //构建 初始化ProductId的请求
                InitProducerIdRequestData requestData = new InitProducerIdRequestData()
                        .setTransactionalId(transactionalId)
                        .setTransactionTimeoutMs(transactionTimeoutMs)
                        .setProducerId(producerIdAndEpoch.producerId)
                        .setProducerEpoch(producerIdAndEpoch.epoch);
                InitProducerIdHandler handler = new InitProducerIdHandler(new InitProducerIdRequest.Builder(requestData),
                        isEpochBump);
                // 把请求加入到请求队列中，等待Sender线程发送
                enqueueRequest(handler);
                return handler.result;
            }, State.INITIALIZING);
        }

InitProducerIdHandler.handleResponse方法

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxNDMwNjY1_size_16_color_FFFFFF_t_70_pic_center 2]

### 开始事务请求 ###

这里不会发起任何请求，只是判断TransactionManager可用性以及当前事务状态是不是ERROR，并把当前事务状态转为IN\_TRANSACTION

public synchronized void beginTransaction() { 
            // 确认TransactionManager不为null
            ensureTransactional();
            //当前状态不为ABORTABLE_ERROR、FATAL_ERROR状态
            maybeFailWithError();
            //把事务状态转为IN_TRANSACTION
            transitionTo(State.IN_TRANSACTION);
        }

### 事务分区上传请求 ###

我们之前讲过，Producer发送消息都是先发送到RecordAccumulator.append，然后到达阀值（大小与时间），Sender线程会把满足条件的批消息发送给broker，在doSend中会判断当前是否要加入到newPartitionsInTransaction中，以备后续上传给TC

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxNDMwNjY1_size_16_color_FFFFFF_t_70_pic_center 3]

public synchronized void maybeAddPartitionToTransaction(TopicPartition topicPartition) { 
            //检查partitionsInTransaction 、newPartitionsInTransaction 、pendingPartitionsInTransaction 中是否存在该topicPartition
            if (isPartitionAdded(topicPartition) || isPartitionPendingAdd(topicPartition))
                //如果已经上传过这个分区，或者正在上传这个分区，那么直接返回
                return;
            log.debug("Begin adding new partition {} to transaction", topicPartition);
            topicPartitionBookkeeper.addPartition(topicPartition);
            //添加到需要上传的集合
            newPartitionsInTransaction.add(topicPartition);
        }

可以看上图，如果`result.batchIsFull || result.newBatchCreated`则唤醒sender线程发送消息

sender每次发送都会调用`runOnce`方法，在发送之前首先会调用`maybeSendAndPollTransactionalRequest`,里面只有`pendingRequests`里的请求为null时或者`TxnRequestHandler instanceof EndTxnHandler &&accumulator有已经完成的待发数据`

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxNDMwNjY1_size_16_color_FFFFFF_t_70_pic_center 4]

则会发送RequestHandler相关的请求。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxNDMwNjY1_size_16_color_FFFFFF_t_70_pic_center 5]

RequestHandler 请求是有优先级关系的（上面表格中有），比如会先发送`事务寻找TC服务地址请求`\->`事务初始化请求`\->…请参考表格，这样就可以发送到`分区上传请求、事务消费位置添加请求、事务消费者位置提交请求`等，等这些都处理完 才会发送Product 产生的消息数据。

### 事务消费位置添加请求 ###

public synchronized TransactionalRequestResult sendOffsetsToTransaction(final Map<TopicPartition, OffsetAndMetadata> offsets, final ConsumerGroupMetadata groupMetadata) { 
            //确定事务可用
            ensureTransactional();
            //失败检查
            maybeFailWithError();
            if (currentState != State.IN_TRANSACTION)
                throw new KafkaException("Cannot send offsets to transaction either because the producer is not in an " +
                        "active transaction");
    
            log.debug("Begin adding offsets {} for consumer group {} to transaction", offsets, groupMetadata);
            //构建 事务消费位置添加 请求
            AddOffsetsToTxnRequest.Builder builder = new AddOffsetsToTxnRequest.Builder(
                new AddOffsetsToTxnRequestData()
                    .setTransactionalId(transactionalId)
                    .setProducerId(producerIdAndEpoch.producerId)
                    .setProducerEpoch(producerIdAndEpoch.epoch)
                    .setGroupId(groupMetadata.groupId())
            );
            AddOffsetsToTxnHandler handler = new AddOffsetsToTxnHandler(builder, offsets, groupMetadata);
            //加入到请求队列
            enqueueRequest(handler);
            return handler.result;
        }

### 事务消费位置提交请求 ###

private TxnOffsetCommitHandler txnOffsetCommitHandler(TransactionalRequestResult result, Map<TopicPartition, OffsetAndMetadata> offsets, ConsumerGroupMetadata groupMetadata) { 
            //遍历所有Partition，把每个Partition中的offset记录到pendingTxnOffsetCommits中
            for (Map.Entry<TopicPartition, OffsetAndMetadata> entry : offsets.entrySet()) { 
                OffsetAndMetadata offsetAndMetadata = entry.getValue();
                CommittedOffset committedOffset = new CommittedOffset(offsetAndMetadata.offset(),
                        offsetAndMetadata.metadata(), offsetAndMetadata.leaderEpoch());
                pendingTxnOffsetCommits.put(entry.getKey(), committedOffset);
            }
            //构建事务消费位置提交请求
            final TxnOffsetCommitRequest.Builder builder =
                new TxnOffsetCommitRequest.Builder(transactionalId,
                    groupMetadata.groupId(),
                    producerIdAndEpoch.producerId,
                    producerIdAndEpoch.epoch,
                    pendingTxnOffsetCommits,
                    groupMetadata.memberId(),
                    groupMetadata.generationId(),
                    groupMetadata.groupInstanceId(),
                    autoDowngradeTxnCommit
                );
      			// 构建 TxnOffsetCommitHandler
            return new TxnOffsetCommitHandler(result, builder);
        }

如果NetworkClient\#poll 执行完成后，会处理response。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxNDMwNjY1_size_16_color_FFFFFF_t_70_pic_center 6]

`completeResponses(response)`里会调用`response.onComplete()`方法,最后`onComplete()`调用`handleResponse(response)`，这里会构建TxnOffsetCommitHandler添加到`penggingRequests`

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxNDMwNjY1_size_16_color_FFFFFF_t_70_pic_center 7]

### 事务提交或回滚请求 ###

//开始事务提交
    public synchronized TransactionalRequestResult beginCommit() { 
            return handleCachedTransactionRequestResult(() -> { 
                maybeFailWithError();
                transitionTo(State.COMMITTING_TRANSACTION);
                return beginCompletingTransaction(TransactionResult.COMMIT);
            }, State.COMMITTING_TRANSACTION);
        }
    //开始事务回滚
    public synchronized TransactionalRequestResult beginAbort() { 
            return handleCachedTransactionRequestResult(() -> { 
                if (currentState != State.ABORTABLE_ERROR)
                    maybeFailWithError();
                transitionTo(State.ABORTING_TRANSACTION);
    
                // We're aborting the transaction, so there should be no need to add new partitions
                newPartitionsInTransaction.clear();
                return beginCompletingTransaction(TransactionResult.ABORT);
            }, State.ABORTING_TRANSACTION);
        }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxNDMwNjY1_size_16_color_FFFFFF_t_70_pic_center 8]

这里如果检测到没有发生错误，则构建事务结束请求，加入到请求.

## 整体流程图 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxNDMwNjY1_size_16_color_FFFFFF_t_70_pic_center 9]

该流程图只是有助于理解事务的流程，以及request、handler等对应关系，不过要更好的理解这块代码，还希望读者可以认真看一遍代码.

github地址：https://github.com/789489498/kafka-trunk

有想一起分析源码的小伙伴可以公众号私信我，咱们共同进步。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxNDMwNjY1_size_16_color_FFFFFF_t_70_pic_center]: /images/20221005/0efc252a9a7743b6b74604815aa2d9e0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxNDMwNjY1_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221005/375791f311a847eba08ac9f92e6c8ea3.png
[20210625013434551.png_pic_center]: /images/20221005/fc17decd43fc448085ad1c0a68821f1f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxNDMwNjY1_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221005/2ddcdf34a2844091a7c55d416a01cf8a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxNDMwNjY1_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221005/1b24d29aa4204612a723f8944a39fc85.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxNDMwNjY1_size_16_color_FFFFFF_t_70_pic_center 4]: /images/20221005/78c67e671d4f48c8a62380a2b4ac8891.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxNDMwNjY1_size_16_color_FFFFFF_t_70_pic_center 5]: /images/20221005/cb992198124045019700e5b60acc24aa.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxNDMwNjY1_size_16_color_FFFFFF_t_70_pic_center 6]: /images/20221005/b5b1d88fd69d4d4b911d10cbf37fd231.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxNDMwNjY1_size_16_color_FFFFFF_t_70_pic_center 7]: /images/20221005/f119f06f82e9400ca70e203442cabed5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxNDMwNjY1_size_16_color_FFFFFF_t_70_pic_center 8]: /images/20221005/61954da1f4f64166a063ad2033c21d85.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxNDMwNjY1_size_16_color_FFFFFF_t_70_pic_center 9]: /images/20221005/d6bb30e23e8f45d293fac88b0331befa.png