Page History

...

__transaction_state: schema

Flow

...

— Regular Consumer

...

Group 

• Producer writes to output topic
  • Source Component: TransactionManager (Client)
  • Action: Sends AddPartitionsToTxn("output-A-0") to TxnCoord.
  • State Change: TxnCoord records participants = {output-A-0} in __transaction_state.
• Producer calls sendOffsetsToTransaction("my-group")
  • Source Component: TransactionManager (Client) & TransactionCoordinator (Broker Core)
  • Action: Sends AddOffsetsToTxn("my-group") to TxnCoord.
  • Execution: TxnCoord computes hash("my-group") = (let's say) 7.
  • State Change: Records participants = {output-A-0, __consumer_offsets-7} in __transaction_state .
• Producer commits
  • Source Component: TransactionCoordinator (Broker Core)
  • Execution: Reads participants from __transaction_state and resolves brokers via MetadataCache:
    • output-A-0  — > Broker 3
    • __consumer_offsets-7 ---→ Broker 4
  • Action: Sends WriteTxnMarkers to Broker 3 AND Broker 4.
• Each broker writes a control batch to its log
  • Source Component: KafkaApis & GroupCoordinator (Broker Core)
  • Execution:
    • Broker 3: Appends COMMIT batch to output-A-0 log (makes records visible to read_committed consumers).
    • Broker 4: Appends COMMIT batch to __consumer_offsets-7 log (makes staged offsets visible to GroupCoordinator).

Flow — Share Group (THIS KIP)

• producer.send("output-A", record)
  • Source Component: TransactionManager (Client)
  • Action: Sends AddPartitionsToTxnRequest(["output-A-0"]).
  • State Change: Updates local state to TransactionMetadata.topicPartitions = {output-A-0}.
• producer.sendShareAcknowledgementsToTransaction(acks, meta)
  • Source Component: TransactionManager (Client)  ---→  KafkaApis (Broker Core)
  • Action: Sends TxnShareAcknowledgeRequest to Broker 5 (the SharePartition leader - lets say it is broker 5).
  • Broker Execution: KafkaApis intercepts the request and performs two background operations:
    1. Resolves the state partition: shareCoordinatorPartition = (__share_group_state, partitionFor(groupId)).
    2. Calls txnCoordinator.handleAddPartitionsToTransaction to dynamically register __share_group_state-N in __transaction_state.
  • State Change: TransactionMetadata now tracks both topicPartitions = {output-A-0, __share_group_state-N}.
  • Staging Phase: Only on success, SharePartitionManager stages the transactional acknowledgements, transitioning the internal InFlightState to TX_PENDING.
• producer.commitTransaction()
  • Source Component: Client  -------> TransactionCoordinator(Broker Core)
  • Action: Sends EndTxnRequest(COMMIT).
  • Execution: TxnCoord reads the topic partitions from state and resolves the physical leaders via MetadataCache:
    • output-A-0 --------> Broker 3
    • __share_group_state-N  -----→ Broker 5
  • Action: Dispatches WriteTxnMarkers to both brokers.
• Each broker handles the transaction markers
  • Execution on Broker 3 (Output Side):
    • Appends the COMMIT control batch to the output-A-0 log.
  • Execution on Broker 5 (Share Side):
    • Appends the COMMIT control batch to the __share_group_state-N log.
    • Fires the KIP-1289 hook: Triggers sharePartitionManager.applyTxnMarker(COMMIT).
    • Scans the internal partitionCache, finds the matching TX_PENDING flight states, and commits them to ACKNOWLEDGED.
• Transaction Completion
  • Action: Both brokers return successful ACKs back to TxnCoord.
  • State Change: TxnCoord transitions the state to COMPLETE_COMMIT.
  • Result: The client-side blocking producer.commitTransaction() call returns successfully.

State machine additions:

• New transient state TX_PENDING 
• On WriteTxnMarkers commit: TX_PENDING(ACCEPT) → ACKNOWLEDGED; same for RELEASE and REJECT.
• On WriteTxnMarkers abort: TX_PENDING(*) → back to ACQUIRED (lock continues; consumer can retry the work).

...