Kafka Transaction Protocol APIs

This document specifies the Kafka protocol APIs used for transactions and exactly-once semantics. These APIs enable atomic writes across multiple partitions and exactly-once processing guarantees.

Transaction API Reference

API Key	Name	Purpose
22	InitProducerId	Initialize producer for idempotence/transactions
24	AddPartitionsToTxn	Add partitions to active transaction
25	AddOffsetsToTxn	Add consumer offsets to transaction
26	EndTxn	Commit or abort transaction
27	WriteTxnMarkers	Write transaction markers (internal)
28	TxnOffsetCommit	Commit offsets within transaction
65	DescribeTransactions	Describe active transactions
66	ListTransactions	List transactions on broker

Transaction Protocol Overview

Transaction Lifecycle

Transaction States

InitProducerId API (Key 22)

Overview

The InitProducerId API initializes a producer for idempotent or transactional operation by assigning a unique Producer ID (PID) and epoch.

Version History

Version	Kafka	Key Changes
0	0.11.0	Initial version
2	2.4.0	Flexible versions
3	2.5.0	Producer ID/epoch in request
4	2.6.0	PRODUCER_FENCED error
5	3.9.0	TRANSACTION_ABORTABLE error (KIP-890)
6	4.0.0	2PC support (KIP-939)

Request Schema

InitProducerIdRequest =>
    transactional_id: NULLABLE_STRING
    transaction_timeout_ms: INT32
    producer_id: INT64
    producer_epoch: INT16
    enable2pc: BOOLEAN
    keep_prepared_txn: BOOLEAN

Field	Type	Description
`transactional_id`	NULLABLE_STRING	Transaction ID (null for idempotent only)
`transaction_timeout_ms`	INT32	Transaction timeout
`producer_id`	INT64	Existing PID to recover (-1 for new)
`producer_epoch`	INT16	Existing epoch to recover (-1 for new)
`enable2pc`	BOOLEAN	Enable two-phase commit (v6+)
`keep_prepared_txn`	BOOLEAN	Keep prepared transaction (v6+)

Response Schema

InitProducerIdResponse =>
    throttle_time_ms: INT32
    error_code: INT16
    producer_id: INT64
    producer_epoch: INT16
    ongoing_txn_producer_id: INT64
    ongoing_txn_producer_epoch: INT16

Field	Type	Description
`producer_id`	INT64	Assigned Producer ID
`producer_epoch`	INT16	Producer epoch
`ongoing_txn_producer_id`	INT64	Producer ID for ongoing txn (v6+)
`ongoing_txn_producer_epoch`	INT16	Producer epoch for ongoing txn (v6+)

Producer ID Semantics

Behavioral Contract

Aspect	Guarantee
PID uniqueness	PID is globally unique per broker lifetime
Epoch increment	Epoch increments on each init with same transactional_id
Fencing	Old producers with same transactional_id are fenced
Timeout	Transaction must complete within transaction_timeout_ms

Error Handling

Error Code	Retriable	Cause	Recovery
COORDINATOR_NOT_AVAILABLE	✅	Coordinator unavailable	Retry
NOT_COORDINATOR	✅	Wrong coordinator	FindCoordinator, retry
CONCURRENT_TRANSACTIONS	✅	Another init in progress	Wait, retry
INVALID_TRANSACTION_TIMEOUT	❌	Timeout out of range	Fix timeout
TRANSACTIONAL_ID_AUTHORIZATION_FAILED	❌	No permission	Check ACLs

AddPartitionsToTxn API (Key 24)

Overview

The AddPartitionsToTxn API adds topic partitions to an active transaction before producing to them.

Version History

Version	Kafka	Key Changes
0	0.11.0	Initial version
2	2.1.0	PRODUCER_FENCED error
3	2.4.0	Flexible versions
4	2.8.0	VerifyOnly + batched transactions
5	3.9.0	TRANSACTION_ABORTABLE (KIP-890)

Request Schema

AddPartitionsToTxnRequest =>
    v3_and_below_transactional_id: STRING
    v3_and_below_producer_id: INT64
    v3_and_below_producer_epoch: INT16
    v3_and_below_topics: [Topic]
    transactions: [Transaction]

Topic =>
    name: STRING
    partitions: [INT32]

Transaction =>
    transactional_id: STRING
    producer_id: INT64
    producer_epoch: INT16
    verify_only: BOOLEAN
    topics: [Topic]

Field	Type	Description
`transactional_id`	STRING	Transaction identifier
`producer_id`	INT64	Producer ID from InitProducerId
`producer_epoch`	INT16	Producer epoch
`ongoing_txn_producer_id`	INT64	Producer ID for ongoing txn (v6+)
`ongoing_txn_producer_epoch`	INT16	Producer epoch for ongoing txn (v6+)
`topics`	ARRAY	Partitions to add

Response Schema

AddPartitionsToTxnResponse =>
    throttle_time_ms: INT32
    error_code: INT16
    results_by_transaction: [TransactionResult]
    results_by_topic_v3_and_below: [TopicResult]

TopicResult =>
    name: STRING
    results_by_partition: [PartitionResult]

PartitionResult =>
    partition_index: INT32
    partition_error_code: INT16

Behavioral Contract

Aspect	Guarantee
Precondition	Transaction must be ongoing (after InitProducerId)
Idempotence	Adding same partition multiple times is safe
Order	Must add partition before producing to it

Produce Before Add

Producing to a partition before adding it to the transaction results in INVALID_PRODUCER_EPOCH or UNKNOWN_PRODUCER_ID error.

AddOffsetsToTxn API (Key 25)

Overview

The AddOffsetsToTxn API adds a consumer group's offsets to an active transaction, enabling exactly-once consume-transform-produce patterns.

Version History

Version	Kafka	Key Changes
0	0.11.0	Initial version
2	2.1.0	PRODUCER_FENCED error
3	2.4.0	Flexible versions
4	3.9.0	TRANSACTION_ABORTABLE (KIP-890)

Request Schema

AddOffsetsToTxnRequest =>
    transactional_id: STRING
    producer_id: INT64
    producer_epoch: INT16
    group_id: STRING

Field	Type	Description
`group_id`	STRING	Consumer group for offset commit

Response Schema

AddOffsetsToTxnResponse =>
    throttle_time_ms: INT32
    error_code: INT16

Behavioral Contract

Aspect	Guarantee
Coordination	Offsets committed atomically with transaction
Group ownership	Does not require group membership
Prerequisite	Must call before TxnOffsetCommit

TxnOffsetCommit API (Key 28)

Overview

The TxnOffsetCommit API commits consumer offsets as part of a transaction.

Version History

Version	Kafka	Key Changes
0	0.11.0	Initial version
2	2.1.0	Committed leader epoch
3	2.4.0	Flexible versions + member fields
4	3.9.0	TRANSACTION_ABORTABLE (KIP-890)
5	4.0.0	Txn V2 add-offsets path (KIP-890 part 2)

Request Schema

TxnOffsetCommitRequest =>
    transactional_id: STRING
    group_id: STRING
    producer_id: INT64
    producer_epoch: INT16
    generation_id: INT32
    member_id: STRING
    group_instance_id: NULLABLE_STRING
    topics: [Topic]

Topic =>
    name: STRING
    partitions: [Partition]

Partition =>
    partition_index: INT32
    committed_offset: INT64
    committed_leader_epoch: INT32
    committed_metadata: NULLABLE_STRING

Field	Type	Description
`generation_id`	INT32	Consumer group generation (-1 if not member)
`member_id`	STRING	Consumer member ID (empty if not member)
`committed_offset`	INT64	Offset to commit

Response Schema

TxnOffsetCommitResponse =>
    throttle_time_ms: INT32
    topics: [Topic]

Topic =>
    name: STRING
    partitions: [Partition]

Partition =>
    partition_index: INT32
    error_code: INT16

Behavioral Contract

Aspect	Guarantee
Atomicity	Offsets committed/aborted with transaction
Visibility	Offsets visible after transaction commit
Prerequisite	Must call AddOffsetsToTxn first

EndTxn API (Key 26)

Overview

The EndTxn API commits or aborts an active transaction.

Version History

Version	Kafka	Key Changes
0	0.11.0	Initial version
2	2.1.0	PRODUCER_FENCED error
3	2.4.0	Flexible versions
4	3.9.0	TRANSACTION_ABORTABLE (KIP-890)
5	4.0.0	Producer ID/epoch in response

Request Schema

EndTxnRequest =>
    transactional_id: STRING
    producer_id: INT64
    producer_epoch: INT16
    committed: BOOLEAN

Field	Type	Description
`committed`	BOOLEAN	true=COMMIT, false=ABORT

Response Schema

EndTxnResponse =>
    throttle_time_ms: INT32
    error_code: INT16

Commit Process

Behavioral Contract

Aspect	Guarantee
Atomicity	All partitions commit or abort together
Durability	Committed transactions survive broker failures
Visibility	After COMMIT, records visible to read_committed consumers
Abort cleanup	Aborted records filtered by read_committed consumers

Error Handling

Error Code	Retriable	Cause	Recovery
INVALID_PRODUCER_ID_MAPPING	❌	PID not found	Reinitialize
INVALID_PRODUCER_EPOCH	❌	Producer fenced	Close producer
INVALID_TXN_STATE	❌	No active transaction	Start new transaction
COORDINATOR_NOT_AVAILABLE	✅	Coordinator down	Retry
CONCURRENT_TRANSACTIONS	✅	Another end in progress	Wait, retry

WriteTxnMarkers API (Key 27)

Overview

The WriteTxnMarkers API is an internal API used by transaction coordinators to write commit/abort markers to partitions.

Internal API

This API is used internally by the Kafka transaction protocol. Clients should not call this API directly.

Version History

Version	Kafka	Key Changes
0	0.11.0	Initial version (removed in 4.0)
1	2.4.0	Flexible versions (4.0 baseline)
2	4.0.0	TransactionVersion (KIP-1228)

Request Schema

WriteTxnMarkersRequest =>
    markers: [Marker]

Marker =>
    producer_id: INT64
    producer_epoch: INT16
    transaction_result: BOOLEAN
    topics: [Topic]
    coordinator_epoch: INT32
    transaction_version: INT8

Topic =>
    name: STRING
    partition_indexes: [INT32]

Field	Type	Description
`transaction_result`	BOOLEAN	true=COMMIT, false=ABORT
`coordinator_epoch`	INT32	Coordinator epoch for fencing

Response Schema

WriteTxnMarkersResponse =>
    markers: [Marker]

Marker =>
    producer_id: INT64
    topics: [Topic]

Topic =>
    name: STRING
    partitions: [Partition]

Partition =>
    partition_index: INT32
    error_code: INT16

DescribeTransactions API (Key 65)

Overview

The DescribeTransactions API retrieves information about active transactions.

Version History

Version	Kafka	Key Changes
0	3.0.0	Initial version

Request Schema

DescribeTransactionsRequest =>
    transactional_ids: [STRING]

Response Schema

DescribeTransactionsResponse =>
    throttle_time_ms: INT32
    transaction_states: [TransactionState]

TransactionState =>
    error_code: INT16
    transactional_id: STRING
    transaction_state: STRING
    transaction_timeout_ms: INT32
    transaction_start_time_ms: INT64
    producer_id: INT64
    producer_epoch: INT16
    topics: [Topic]

Topic =>
    topic: STRING
    partitions: [INT32]

Field	Type	Description
`transaction_state`	STRING	Current state (Empty, Ongoing, etc.)
`transaction_start_time_ms`	INT64	Transaction start timestamp
`topics`	ARRAY	Partitions in transaction

ListTransactions API (Key 66)

Overview

The ListTransactions API lists transactions on a broker with optional filtering.

Version History

Version	Kafka	Key Changes
0	3.0.0	Initial version
1	3.5.0	Duration filter
2	4.0.0	TransactionalIdPattern (KIP-1152)

Request Schema

ListTransactionsRequest =>
    state_filters: [STRING]
    producer_id_filters: [INT64]
    duration_filter: INT64
    transactional_id_pattern: NULLABLE_STRING

Field	Type	Description
`state_filters`	ARRAY	Filter by state (empty for all)
`producer_id_filters`	ARRAY	Filter by PID (empty for all)
`duration_filter`	INT64	Minimum transaction duration (ms)
`transactional_id_pattern`	NULLABLE_STRING	Regex filter for transactional.id (v2+)

Response Schema

ListTransactionsResponse =>
    throttle_time_ms: INT32
    error_code: INT16
    unknown_state_filters: [STRING]
    transaction_states: [TransactionState]

TransactionState =>
    transactional_id: STRING
    producer_id: INT64
    transaction_state: STRING

Exactly-Once Processing Pattern

Consume-Transform-Produce

Isolation Level Interaction

Consumer Isolation	Behavior
`read_uncommitted`	Sees all records including uncommitted
`read_committed`	Only sees committed records, filters aborted

Transaction Configuration

Producer Configuration

Setting	Default	Description
`transactional.id`	null	Unique transaction identifier
`transaction.timeout.ms`	60000	Maximum transaction duration
`enable.idempotence`	true	Required for transactions

Broker Configuration

Setting	Default	Description
`transaction.state.log.replication.factor`	3	Replication for transaction log
`transaction.state.log.min.isr`	2	Min ISR for transaction log
`transactional.id.expiration.ms`	604800000	Transaction ID expiration
`transaction.max.timeout.ms`	900000	Maximum allowed timeout

Consumer Configuration

Setting	Default	Description
`isolation.level`	read_uncommitted	Transaction isolation

Transaction Best Practices

Producer Guidelines

Practice	Rationale
Use unique transactional.id per producer instance	Prevents conflicts
Keep transactions short	Avoid timeout and resource pressure
Handle ProducerFencedException	Indicates another producer took over
Abort on any error	Clean up partial transactions

Consumer Guidelines

Practice	Rationale
Use read_committed for exactly-once	Filter uncommitted and aborted
Process in batches within transactions	Amortize transaction overhead
Store offsets in transaction	Atomic offset updates

Error Recovery

Protocol Core APIs - Core APIs
Protocol Consumer APIs - Consumer group APIs
Protocol Admin APIs - Administrative APIs
Protocol Errors - Error code reference
Failure Handling - Idempotent producers

Kafka Transaction Protocol APIs

Transaction API Reference

Transaction Protocol Overview

Transaction Lifecycle

Transaction States

InitProducerId API (Key 22)

Overview

Version History

Request Schema

Response Schema

Producer ID Semantics

Behavioral Contract

Error Handling

AddPartitionsToTxn API (Key 24)

Overview

Version History

Request Schema

Response Schema

Behavioral Contract

AddOffsetsToTxn API (Key 25)

Overview

Version History

Request Schema

Response Schema

Behavioral Contract

TxnOffsetCommit API (Key 28)

Overview

Version History

Request Schema

Response Schema

Behavioral Contract

EndTxn API (Key 26)

Overview

Version History

Request Schema

Response Schema

Commit Process

Behavioral Contract

Error Handling

WriteTxnMarkers API (Key 27)

Overview

Version History

Request Schema

Response Schema

DescribeTransactions API (Key 65)

Overview

Version History

Request Schema

Response Schema

ListTransactions API (Key 66)

Overview

Version History

Request Schema

Response Schema

Exactly-Once Processing Pattern

Consume-Transform-Produce

Isolation Level Interaction

Transaction Configuration

Producer Configuration

Broker Configuration

Consumer Configuration

Transaction Best Practices

Producer Guidelines

Consumer Guidelines

Error Recovery

Related Documentation