1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

// 創(chuàng)建 Producer 實(shí)例，并且指定 transaction id
KafkaProducer producer = createKafkaProducer(
  “bootstrap.servers”, “l(fā)ocalhost:9092”,
  “transactional.id”, “my-transactional-id”);

// 初始化事務(wù)，這里會向 TC 服務(wù)申請 producer id
producer.initTransactions();

// 創(chuàng)建 Consumer 實(shí)例，并且訂閱 topic
KafkaConsumer consumer = createKafkaConsumer(
  “bootstrap.servers”, “l(fā)ocalhost:9092”,
  “group.id”, “my-group-id”,
  "isolation.level", "read_committed");
consumer.subscribe(singleton(“inputTopic”));

while (true) {
  ConsumerRecords records = consumer.poll(Long.MAX_VALUE);
  // 開始新的事務(wù)
  producer.beginTransaction();
  for (ConsumerRecord record : records) {
    // 發(fā)送消息到分區(qū)
    producer.send(producerRecord(“outputTopic_1”, record));
    producer.send(producerRecord(“outputTopic_2”, record));
  }
  // 提交 offset
  producer.sendOffsetsToTransaction(currentOffsets(consumer), "my-group-id");  
  // 提交事務(wù)
  producer.commitTransaction();
}

1
2
3
4
5
6
7
8
9
10

private enum State {
    UNINITIALIZED,
    INITIALIZING,
    READY,
    IN_TRANSACTION,
    COMMITTING_TRANSACTION,
    ABORTING_TRANSACTION,
    ABORTABLE_ERROR,
    FATAL_ERROR;
}

1
2
3
4
5
6
7
8
9

class TransactionMetadata(
    val transactionalId: String,      // 事務(wù) id
    var producerId: Long,             // pruducer id
    var producerEpoch: Short,         // producer epoch
    var txnTimeoutMs: Int,            // 事務(wù)超時時間
    var state: TransactionState,      // 事務(wù)當(dāng)前狀態(tài)
    val topicPartitions: mutable.Set[TopicPartition],    // 該事務(wù)涉及到的分區(qū)列表
    @volatile var txnStartTimestamp: Long = -1,          // 事務(wù)開始的時間
    @volatile var txnLastUpdateTimestamp: Long)          // 事務(wù)的更新時間

1
2
3
4
5
6
7
8
9
10
11
12
13

public interface Producer<K, V> extends Closeable {
    // 初始化事務(wù)，包括申請 producer id
    void initTransactions();
    // 開始事務(wù)，這里會更改事務(wù)的本地狀態(tài)
    void beginTransaction() throws ProducerFencedException;
    // 提交消費(fèi)位置， offsets表示每個分區(qū)的消費(fèi)位置， consumerGroupId表示消費(fèi)組的名稱
    void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,
                                  String consumerGroupId) throws ProducerFencedException;
    // 發(fā)送事務(wù)提交請求
    void commitTransaction() throws ProducerFencedException;
    // 發(fā)送事務(wù)回滾請求
    void abortTransaction() throws ProducerFencedException;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

public synchronized TransactionalRequestResult initializeTransactions() {
    // 檢查transaction id是否已經(jīng)設(shè)置
    ensureTransactional();
    // 更改自身狀態(tài)為INITIALIZING
    transitionTo(State.INITIALIZING);
    // 將producer id和epoch都設(shè)為空
    setProducerIdAndEpoch(ProducerIdAndEpoch.NONE);
    // nextSequence在消息發(fā)送中會用到，因?yàn)榘l(fā)送事務(wù)消息要求冪等性，而發(fā)送冪等性的消息是需要設(shè)置sequence的
    this.nextSequence.clear();
    // 構(gòu)建申請produce id請求
    InitProducerIdRequest.Builder builder = new InitProducerIdRequest.Builder(transactionalId, transactionTimeoutMs);
    // InitProducerIdHandler 負(fù)責(zé)處理響應(yīng)
    InitProducerIdHandler handler = new InitProducerIdHandler(builder);
    // 將消息保存到隊列中，等待Sender線程（Producer會有個后臺線程發(fā)送消息）發(fā)送
    enqueueRequest(handler);
    // 返回異步結(jié)果
    return handler.result;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

private class InitProducerIdHandler extends TxnRequestHandler {

    @Override
    public void handleResponse(AbstractResponse response) {
        InitProducerIdResponse initProducerIdResponse = (InitProducerIdResponse) response;
        // 檢查錯誤
        Errors error = initProducerIdResponse.error();
        if (error == Errors.NONE) {
            ProducerIdAndEpoch producerIdAndEpoch = new ProducerIdAndEpoch(initProducerIdResponse.producerId(), initProducerIdResponse.epoch());
            // 保存結(jié)果 producer id和epoch
            setProducerIdAndEpoch(producerIdAndEpoch);
            // 更改狀態(tài)為READY
            transitionTo(State.READY);
            lastError = null;
            // 通知異步結(jié)果已完成
            result.done();
        } else if (error == Errors.NOT_COORDINATOR || error == Errors.COORDINATOR_NOT_AVAILABLE) {
            // 如果TC服務(wù)沒有找到或者剛好掛掉，那么生成FindCoordinatorRequest請求，等待發(fā)送
            lookupCoordinator(FindCoordinatorRequest.CoordinatorType.TRANSACTION, transactionalId);
            // 并且將自身請求也放入隊列，等待發(fā)送
            reenqueue();
        } else if (error == Errors.COORDINATOR_LOAD_IN_PROGRESS || error == Errors.CONCURRENT_TRANSACTIONS) {
            // 如果TC服務(wù)正在啟動中，那么加入隊列，等待發(fā)送
            reenqueue();
        } else if (error == Errors.TRANSACTIONAL_ID_AUTHORIZATION_FAILED) {
            // 如果發(fā)生權(quán)限問題，那么認(rèn)為進(jìn)入錯誤狀態(tài)
            fatalError(error.exception());
        } else {
            fatalError(new KafkaException("Unexpected error in InitProducerIdResponse; " + error.message()));
        }
    }
}

1
2
3
4
5
6
7
8

public synchronized void beginTransaction() {
    // 檢查 transaction id
    ensureTransactional();
    // 檢查之前響應(yīng)是否出錯
    maybeFailWithError();
    // 更改狀態(tài)
    transitionTo(State.IN_TRANSACTION);
}

1
2
3
4
5
6
7
8
9
10
11
12

public class KafkaProducer<K, V> implements Producer<K, V> {
    private Future<RecordMetadata> doSend(ProducerRecord<K, V> record, Callback callback) {
        .......
        int partition = partition(record, serializedKey, serializedValue, cluster);
        tp = new TopicPartition(record.topic(), partition);
        ......
        if (transactionManager != null && transactionManager.isTransactional())
            // 如果開啟了事務(wù)，那么就先將分區(qū)保存在 transactionManager
            transactionManager.maybeAddPartitionToTransaction(tp);
        ......
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

public class TransactionManager {
    // 新增的上傳分區(qū)集合
    private final Set<TopicPartition> newPartitionsInTransaction;
    // 本次事務(wù)已經(jīng)上傳的分區(qū)集合
    private final Set<TopicPartition> partitionsInTransaction;
    // 本次事務(wù)涉及到的所有分區(qū)集合
    private final Set<TopicPartition> pendingPartitionsInTransaction;

    public synchronized void maybeAddPartitionToTransaction(TopicPartition topicPartition) {
        // 檢查事務(wù)狀態(tài)必須為IN_TRANSACTION
        failIfNotReadyForSend();
        // 如果已經(jīng)上傳過這個分區(qū)，或者正在上傳這個分區(qū)，那么直接返回
        if (isPartitionAdded(topicPartition) || isPartitionPendingAdd(topicPartition))
            return;

        log.debug("Begin adding new partition {} to transaction", topicPartition);
        // 添加到需要上次的集合
        newPartitionsInTransaction.add(topicPartition);
    }

    // 檢查是否這個分區(qū)已經(jīng)上傳過了
    synchronized boolean isPartitionAdded(TopicPartition partition) {
        return partitionsInTransaction.contains(partition);
    }

    // 檢查是否這個分區(qū)正在上傳中
    synchronized boolean isPartitionPendingAdd(TopicPartition partition) {
        return newPartitionsInTransaction.contains(partition) || pendingPartitionsInTransaction.contains(partition);
    }    
}

1
2
3
4
5
6
7
8
9
10
11
12
13

public class TransactionManager {

    private synchronized TxnRequestHandler addPartitionsToTransactionHandler() {
        // 將新增的分區(qū)，添加到 pendingPartitionsInTransaction 集合
        pendingPartitionsInTransaction.addAll(newPartitionsInTransaction);
        // 清空新增的分區(qū)集合
        newPartitionsInTransaction.clear();
        // 構(gòu)建 AddPartitionsToTxnRequest 請求
        AddPartitionsToTxnRequest.Builder builder = new AddPartitionsToTxnRequest.Builder(transactionalId,
                producerIdAndEpoch.producerId, producerIdAndEpoch.epoch, new ArrayList<>(pendingPartitionsInTransaction));
        return new AddPartitionsToTxnHandler(builder);
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

private class AddPartitionsToTxnHandler extends TxnRequestHandler {

    @Override
    public void handleResponse(AbstractResponse response) {
        AddPartitionsToTxnResponse addPartitionsToTxnResponse = (AddPartitionsToTxnResponse) response;
        Map<TopicPartition, Errors> errors = addPartitionsToTxnResponse.errors();
        boolean hasPartitionErrors = false;
        Set<String> unauthorizedTopics = new HashSet<>();
        retryBackoffMs = TransactionManager.this.retryBackoffMs;

        for (Map.Entry<TopicPartition, Errors> topicPartitionErrorEntry : errors.entrySet()) {
            // 檢查每個分區(qū)的響應(yīng)錯誤
            .....
        }

        Set<TopicPartition> partitions = errors.keySet();
        // 因?yàn)檫@些分區(qū)已經(jīng)有響應(yīng)了，即使錯誤也需要從集合中刪除
        pendingPartitionsInTransaction.removeAll(partitions);

        if (!unauthorizedTopics.isEmpty()) {
            abortableError(new TopicAuthorizationException(unauthorizedTopics));
        } else if (hasPartitionErrors) {
            abortableError(new KafkaException("Could not add partitions to transaction due to errors: " + errors));
        } else {
            log.debug("Successfully added partitions {} to transaction", partitions);
            // 將這些成功響應(yīng)的分區(qū)，添加到 partitionsInTransaction集合
            partitionsInTransaction.addAll(partitions);
            transactionStarted = true;
            // 通知結(jié)果成功
            result.done();
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

public class TransactionManager {
    public synchronized TransactionalRequestResult sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets, String consumerGroupId) {
        // 保證transaction id
        ensureTransactional();
        // 檢查之前的響應(yīng)錯誤
        maybeFailWithError();
        // 只有 IN_TRANSACTION 狀態(tài)才可以發(fā)送這種類型請求
        if (currentState != State.IN_TRANSACTION)
            throw new KafkaException("...");
        // 構(gòu)建請求
        AddOffsetsToTxnRequest.Builder builder = new AddOffsetsToTxnRequest.Builder(transactionalId, producerIdAndEpoch.producerId, producerIdAndEpoch.epoch, consumerGroupId);
        // 構(gòu)建處理器
        AddOffsetsToTxnHandler handler = new AddOffsetsToTxnHandler(builder, offsets);
        // 添加到隊列
        enqueueRequest(handler);
        return handler.result;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

public synchronized TransactionalRequestResult beginCommit() {
    ensureTransactional();
    maybeFailWithError();
    // 更改狀態(tài)為 COMMITTING_TRANSACTION
    transitionTo(State.COMMITTING_TRANSACTION);
    // 調(diào)用 beginCompletingTransaction 方法發(fā)送請求
    return beginCompletingTransaction(TransactionResult.COMMIT);
}

public synchronized TransactionalRequestResult beginAbort() {
    ensureTransactional();
    // 更改狀態(tài)為 ABORTABLE_ERROR
    if (currentState != State.ABORTABLE_ERROR)
        maybeFailWithError();
    transitionTo(State.ABORTING_TRANSACTION);

    // 清空分區(qū)集合
    newPartitionsInTransaction.clear();
    // 發(fā)送請求
    return beginCompletingTransaction(TransactionResult.ABORT);
}

private TransactionalRequestResult beginCompletingTransaction(TransactionResult transactionResult) {
    // 如果還有分區(qū)沒有上傳，那么生成請求放進(jìn)隊列
    if (!newPartitionsInTransaction.isEmpty())
        enqueueRezquest(addPartitionsToTransactionHandler());
    // 構(gòu)建請求
    EndTxnRequest.Builder builder = new EndTxnRequest.Builder(transactionalId, producerIdAndEpoch.producerId, producerIdAndEpoch.epoch, transactionResult);
    // 構(gòu)建處理器
    EndTxnHandler handler = new EndTxnHandler(builder);
    enqueueRequest(handler);
    return handler.result;
}

1
2
3
4
5
6
7
8
9

class TransactionStateManager(...) {
  // key值為 partition id，value為 TxnMetadataCacheEntry對象
  private val transactionMetadataCache: mutable.Map[Int, TxnMetadataCacheEntry] = mutable.Map()
}

// metadataPerTransactionalId 參數(shù)是Pool類型，可以看成是Map
// key值為transaction id， value為元數(shù)據(jù)
private[transaction] case class TxnMetadataCacheEntry(coordinatorEpoch: Int,
                                                      metadataPerTransactionalId: Pool[String, TransactionMetadata])

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

class TransactionStateManager {

  def appendTransactionToLog(transactionalId: String,
                             coordinatorEpoch: Int,
                             newMetadata: TxnTransitMetadata,  // 新的元數(shù)據(jù)
                             responseCallback: Errors => Unit, // 回調(diào)函數(shù)
                             retryOnError: Errors => Boolean = _ => false): Unit = {
    // 生成record的key
    val keyBytes = TransactionLog.keyToBytes(transactionalId)
    // 生成record的calue
    val valueBytes = TransactionLog.valueToBytes(newMetadata)
    // 生成record
    val records = MemoryRecords.withRecords(TransactionLog.EnforcedCompressionType, new SimpleRecord(timestamp, keyBytes, valueBytes))
    val topicPartition = new TopicPartition(Topic.TRANSACTION_STATE_TOPIC_NAME, partitionFor(transactionalId))
    val recordsPerPartition = Map(topicPartition -> records)      

    // 當(dāng)持久化完成后，會調(diào)用這個函數(shù)，更新transactionMetadataCache集合的元數(shù)據(jù)
    def updateCacheCallback(responseStatus: collection.Map[TopicPartition, PartitionResponse]): Unit = {
        // 檢查持久化是否出錯
        .....
        // 更改元數(shù)據(jù)
        metadata.completeTransitionTo(newMetadata)
        // 執(zhí)行回調(diào)函數(shù)
        responseCallback(responseError)
    }

    ......

    // 持久化recod到topic里
    replicaManager.appendRecords(
                newMetadata.txnTimeoutMs.toLong,
                TransactionLog.EnforcedRequiredAcks,
                internalTopicsAllowed = true,
                isFromClient = false,
                recordsPerPartition,
                updateCacheCallback,   // 持久化完成后，會調(diào)用這個函數(shù)
                delayedProduceLock = Some(stateLock.readLock))
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

def handleEndTransaction(transactionalId: String,
                         producerId: Long,
                         producerEpoch: Short,
                         txnMarkerResult: TransactionResult,
                         responseCallback: EndTxnCallback): Unit = {
    // 獲取元數(shù)據(jù)，更改狀態(tài)
    val preAppendResult: ApiResult[(Int, TxnTransitMetadata)] = txnManager.getTransactionState(transactionalId).right.flatMap {
      case None =>
        // 如果元數(shù)據(jù)不存在，說明存在問題
        Left(Errors.INVALID_PRODUCER_ID_MAPPING)
      case Some(epochAndTxnMetadata) =>
        val txnMetadata = epochAndTxnMetadata.transactionMetadata
        val coordinatorEpoch = epochAndTxnMetadata.coordinatorEpoch
        ......
        txnMetadata.state match {
            // 必須是Ongoing狀態(tài)，如果是別的狀態(tài)，就會報錯
            case Ongoing =>
              // 根據(jù)發(fā)送的消息類型，查看是事務(wù)提交還是回滾，來決定接下來的狀態(tài)
              val nextState = if (txnMarkerResult == TransactionResult.COMMIT)
                PrepareCommit
              else
                PrepareAbort
              // 調(diào)用 prepareAbortOrCommit 來更新狀態(tài)
              Right(coordinatorEpoch, txnMetadata.prepareAbortOrCommit(nextState, time.milliseconds()))
        }
    }

    // 檢查上一步的結(jié)果
    preAppendResult match {
      case Left(err) =>
        responseCallback(err)
      case Right((coordinatorEpoch, newMetadata)) =>

        // 定義回調(diào)函數(shù)，用于發(fā)送請求到分區(qū)
        def sendTxnMarkersCallback(error: Errors): Unit = {
          if (error == Errors.NONE) {
              ...... // 檢查狀態(tài)，只有是PrepareCommit或PrepareAbort，才能繼續(xù)執(zhí)行
            txnMetadata.state match {
              case PrepareCommit =>
                // 更改狀態(tài)為 CompleteCommit
                txnMetadata.prepareComplete(time.milliseconds())
              case PrepareAbort =>
                // 更改狀態(tài)為 CompleteAbort
                txnMetadata.prepareComplete(time.milliseconds())
            }
          }
          // 檢查更改狀態(tài)的結(jié)果
          preSendResult match {
            case Left(err) =>
              responseCallback(err)
            case Right((txnMetadata, newPreSendMetadata)) =>
              // 向客戶端發(fā)送成功響應(yīng)
              responseCallback(Errors.NONE)
              // 通過txnMarkerChannelManager發(fā)送請求到分區(qū)
              txnMarkerChannelManager.addTxnMarkersToSend(transactionalId, coordinatorEpoch, txnMarkerResult, txnMetadata, newPreSendMetadata)
          }
        }

        // 持久化元數(shù)據(jù)，然后調(diào)用sendTxnMarkersCallback函數(shù)，發(fā)送客戶端的響應(yīng)和發(fā)送請求到分區(qū)
        txnManager.appendTransactionToLog(transactionalId, coordinatorEpoch, newMetadata, sendTxnMarkersCallback)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

private[transaction] class DelayedTxnMarker(txnMetadata: TransactionMetadata,
                                           completionCallback: Errors => Unit,
                                           lock: Lock)
  extends DelayedOperation(TimeUnit.DAYS.toMillis(100 * 365), Some(lock)) {

  override def tryComplete(): Boolean = {
    txnMetadata.inLock {
      // 當(dāng)每收到一個分區(qū)的響應(yīng)后，就會從元數(shù)據(jù)中刪除掉。
      // 直到分區(qū)列表為空，就說明所有分區(qū)都已經(jīng)成功響應(yīng)
      if (txnMetadata.topicPartitions.isEmpty)
        forceComplete()
      else false
    }
  }

  override def onExpiration(): Unit = {
    // this should never happen
    throw new IllegalStateException(s"Delayed write txn marker operation for metadata $txnMetadata has timed out, this should never happen.")
  }

  // TODO: if we will always return NONE upon completion, we can remove the error code in the param
  override def onComplete(): Unit = completionCallback(Errors.NONE)
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

def addTxnMarkersToSend(transactionalId: String,
                        coordinatorEpoch: Int,
                        txnResult: TransactionResult,
                        txnMetadata: TransactionMetadata,
                        newMetadata: TxnTransitMetadata): Unit = {
  // 定義延遲任務(wù)的回調(diào)函數(shù)
  def appendToLogCallback(error: Errors): Unit = {
    // 檢查錯誤
    error match {
      case Errors.NONE =>
        // 檢查狀態(tài)
        txnStateManager.getTransactionState(transactionalId) match {
          case Right(Some(epochAndMetadata)) =>
            if (epochAndMetadata.coordinatorEpoch == coordinatorEpoch) {
               // 持久化事務(wù)成功消息TxnLogAppend，寫入到事務(wù) topic
              tryAppendToLog(TxnLogAppend(transactionalId, coordinatorEpoch, txnMetadata, newMetadata))
            }
        }
    }
  }

  // 實(shí)例化延遲任務(wù)
  val delayedTxnMarker = new DelayedTxnMarker(txnMetadata, appendToLogCallback, txnStateManager.stateReadLock)
  // 等待執(zhí)行
  txnMarkerPurgatory.tryCompleteElseWatch(delayedTxnMarker, Seq(transactionalId))
  // 將請求放進(jìn)隊列里
  addTxnMarkersToBrokerQueue(transactionalId, txnMetadata.producerId, txnMetadata.producerEpoch, txnResult, coordinatorEpoch, txnMetadata.topicPartitions.toSet)
}