14 · Transactions & Exactly-Once Semantics

Source: Apache Kafka 4.4.0-SNAPSHOT (git 04bfe7d, 2026-06-15), KRaft mode. Derived from source code, not copied from official documentation.

Kafka builds exactly-once semantics (EOS) in two layers that compose. The lower layer is the idempotent producer: every record batch carries a (producerId, producerEpoch, baseSequence) triple, and the partition leader's ProducerStateManager deduplicates retries and enforces a strictly increasing per-partition sequence, giving exactly-once-in-order delivery to a single partition. The upper layer is transactions (KIP-98): a stable transactional.id is mapped to a producer id with epoch fencing of zombies; a TransactionCoordinator drives a two-phase commit over an internal __transaction_state log and writes transaction markers (control records) into every partition the transaction touched. Consumers reading at read_committed use the Last Stable Offset and a per-segment aborted-transaction index to filter out aborted data. This chapter traces the whole path end to end, grounded in the source.

Role & responsibilities

The transaction/EOS subsystem must deliver three guarantees:

• Idempotence, a producer retrying a batch (e.g. after a network hiccup) must not create a duplicate. This is enforced per (producerId, epoch) per partition by the leader's sequence-number check and a 5-batch dedup cache.
• Atomic multi-partition writes, a set of writes across many partitions (and, for consume-transform-produce pipelines, the consumer offset commit) either all become visible or none do. This is the job of the TransactionCoordinator + __transaction_state log + transaction markers.
• Zombie fencing, a stale producer instance (one that crashed and was replaced) must be prevented from writing into a transaction owned by its successor. This is achieved with the producer epoch: InitProducerId bumps the epoch, and the leader rejects any batch/marker with a lower epoch.

Together these implement consume-transform-produce EOS: a Kafka Streams (see Kafka Streams Architecture) or hand-written application reads from input partitions, produces derived records, and commits its input offsets, all inside one transaction. The offsets are written to __consumer_offsets by the group coordinator but only become visible to other consumers when the transaction commits.

Where it lives in the code

Core concepts & terminology

Producer ID (PID)

A cluster-unique int64 stamped in every batch produced by an idempotent/transactional producer. Allocated in blocks of 1000 (ProducerIdsBlock.PRODUCER_ID_BLOCK_SIZE = 1000, server-common/.../ProducerIdsBlock.java:30).

Producer epoch

An int16 generation counter for a PID, used to fence zombies. Bumped on InitProducerId and (under Transaction V2) on every commit/abort.

Sequence number

A per-partition int32 that starts at 0 and increments by the number of records in each batch; the leader requires nextSeq == lastSeq + 1.

transactional.id

A stable, user-supplied string identifying a logical producer across restarts. It maps deterministically to one __transaction_state partition (and thus one coordinator).

Transaction marker

A control record (COMMIT or ABORT) the coordinator writes into each data partition to atomically end a transaction. Carries the coordinator epoch for fencing.

Last Stable Offset (LSO)

min(highWatermark, firstUnstableOffset), the highest offset below which there are no in-flight (undecided) transactions. read_committed consumers may not read past it.

Coordinator epoch

The leader epoch of the __transaction_state partition; rises on every coordinator failover and fences stale marker writes / log appends.

Transaction Version (TV)

A cluster feature flag (transaction.version). TV_0/TV_1 are the legacy protocol; TV_2 (KIP-890) bumps the epoch per transaction so each transaction is uniquely identified by (pid, epoch) (server-common/.../TransactionVersion.java:24-31).

The idempotent producer

PID allocation: controller block-allocation

PIDs are allocated by the active KRaft controller in monotonic blocks, never one at a time, so that allocation is cheap and survives controller failover via the metadata log. ProducerIdControlManager.generateNextProducerId takes the next start id, builds a ProducerIdsBlock of size 1000, and emits a ProducerIdsRecord whose nextProducerId field advances past the block; the new value is committed to the metadata log before the block is handed out (metadata/.../ProducerIdControlManager.java:84-101). On replay, the manager asserts strict monotonicity, a replayed nextProducerId must exceed the current block's first id (:111-120), which is the persistent invariant that no PID is ever reused.

On the broker side, RPCProducerIdManager caches the current block and serves generateProducerId() lock-free by currentProducerIdBlock.get().claimNextId(), an atomic getAndIncrement on the block's counter (server-common/.../ProducerIdsBlock.java:50-56). When 90% of the block is consumed (PID_PREFETCH_THRESHOLD = 0.90) it asynchronously requests the next block from the controller via AllocateProducerIds, gated by a single-flight requestInFlight CAS; if the block runs dry before the next arrives it throws a retriable COORDINATOR_LOAD_IN_PROGRESS rather than blocking (transaction-coordinator/.../RPCProducerIdManager.java:80-109).

Sequence validation & deduplication on the leader

The partition leader's per-PID state lives in ProducerStateEntry, which keeps a bounded deque of the last 5 batches (ProducerStateEntry.NUM_BATCHES_TO_RETAIN = 5, storage/.../ProducerStateEntry.java:35). Each entry records producerEpoch, coordinatorEpoch, lastTimestamp, an optional currentTxnFirstOffset, and for each retained batch a BatchMetadata(lastSeq, lastOffset, offsetDelta, timestamp) (storage/.../BatchMetadata.java:21).

On a client produce, UnifiedLog.analyzeAndValidateProducerState first checks for an exact duplicate: for each batch with a producer id it asks ProducerStateEntry.findDuplicateBatch, which (for the matching epoch) looks for a cached batch with the identical [baseSequence, lastSequence] range (storage/.../ProducerStateEntry.java:128-137). If found, the append short-circuits, no data is written, and the leader returns the original offset/timestamp of the already-stored batch to the client, so a retry looks like a success (storage/.../UnifiedLog.java:1247-1254).

If it is not a duplicate, ProducerAppendInfo.checkSequence enforces ordering (storage/.../ProducerAppendInfo.java:156-194):

• Same epoch: require inSequence(currentLastSeq, appendFirstSeq), i.e. appendFirstSeq == currentLastSeq + 1 (with wraparound at Integer.MAX_VALUE → 0). A gap throws OutOfOrderSequenceException (:196-198, :188-192).
• New epoch with non-zero first sequence and a known prior epoch throws OutOfOrderSequenceException, a fresh epoch must restart at sequence 0 (:169-176).
• If all prior records for the PID were deleted (epoch reset to NO_PRODUCER_EPOCH) the leader accepts any sequence (:188), which is why a long-idle producer can hit UNKNOWN_PRODUCER_ID and recover.

Epoch fencing at the leader

ProducerAppendInfo.checkProducerEpoch rejects stale writers. The comparison depends on the transaction version of the incoming marker: for TV_2 a marker is valid only if markerEpoch > currentEpoch (strictly greater, because TV_2 bumps the epoch before writing the marker); for legacy TV_0/TV_1 it accepts markerEpoch >= currentEpoch (storage/.../ProducerAppendInfo.java:117-153). A client data batch whose epoch is below the leader's last-seen epoch throws InvalidProducerEpochException (a recoverable error, the client aborts and retries), not the older fatal ProducerFenced (:145-152).

Transactions: data structures

In-memory: TransactionMetadata

Each transactional.id the coordinator owns has one TransactionMetadata, guarded by its own ReentrantLock exposed via inLock(...) (transaction-coordinator/.../TransactionMetadata.java:62, 111-118). Key fields:

A prospective transition is captured in an immutable TxnTransitMetadata record (transaction-coordinator/.../TxnTransitMetadata.java:27-41). The two-step protocol is prepareXxx() → set pendingState and return a TxnTransitMetadata, then after the log write succeeds, completeTransitionTo(transit) validates and atomically overwrites the live fields (TransactionMetadata.java:318-446). completeTransitionTo re-checks state validity, epoch progression, timestamp monotonicity and that the old partition set is a subset of the new, corrupt transitions throw IllegalStateException marked FATAL.

Persistent: the __transaction_state log

Transaction metadata is durably stored in the internal compacted topic __transaction_state. Its topic config, min.insync.replicas=2 (from transaction.state.log.min.isr), uncompressed, cleanup.policy=compact, unclean.leader.election=false, plus segment size, is set in TransactionCoordinator.transactionStateTopicConfigs (TransactionCoordinator.scala:1014-1022); the partition count (50 by default) and replication factor (3) are not set there but come from transaction.state.log.num.partitions / .replication.factor (TransactionLogConfig.java:32, 40) and are applied when the internal topic is created. The record key is the transactional id, prefixed with a coordinator record type id; the value is a versioned TransactionLogValue (transaction-coordinator/.../TransactionLog.java:49-95).

Because the topic is compacted, only the latest record per transactional id is retained, which is sufficient to rebuild the coordinator's in-memory state after failover. TransactionLog.read decodes a record into a TxnRecord (id + metadata), a TxnTombstone, or an unknown-version marker that is logged and skipped (TransactionLog.java:118-164).

The transaction state machine

TransactionState defines 8 states and a VALID_PREVIOUS_STATES map that prepareTransitionTo enforces (transaction-coordinator/.../TransactionState.java:31-103):

Notable edges from the code:

• EMPTY, COMPLETE_COMMIT, COMPLETE_ABORT can all transition to ONGOING (a new transaction reuses the metadata) and, under TV_2, directly to PREPARE_ABORT, because a client uncertain of server state may send EndTxn(abort) with no partitions added (TransactionState.java:94-103; logic in TransactionCoordinator.endTransaction, :894-899).
• PREPARE_EPOCH_FENCE is reachable only from ONGOING and is used when a new InitProducerId arrives while a transaction is live, or when the background timeout reaper fires: the coordinator bumps the epoch, aborts the dangling transaction, then returns CONCURRENT_TRANSACTIONS to force the client to retry (TransactionCoordinator.scala:281-287, 190-207).
• DEAD is a transient state during transactional-id expiration; on success the metadata is removed from the cache, never persisted as DEAD (TransactionMetadata.java:421-426).

The protocol flow, step by step

1 · InitProducerId, bind, fence, abort dangling

TransactionCoordinator.handleInitProducerId (TransactionCoordinator.scala:117-226) routes on the transactional id:

• Null id → a pure idempotent (non-transactional) producer: blindly allocate a fresh PID with epoch 0 and return (:125-132).
• Real id → look up or create the metadata (a new metadata gets a freshly generated PID, epoch -1, state EMPTY, :154-166), then prepareInitProducerIdTransit decides the transition under the metadata lock.

prepareInitProducerIdTransit (:228-296) is where fencing and recovery live. If the state is EMPTY/COMPLETE_* it bumps the epoch via prepareIncrementProducerEpoch; if the epoch is exhausted it instead rotates to a new PID via prepareProducerIdRotation (:264-271). If a transaction is ONGOING, it returns prepareFenceProducerEpoch(), the coordinator will abort that transaction first and reply CONCURRENT_TRANSACTIONS so the client retries (:281-287, 190-207). An expectedProducerIdAndEpoch that does not match (and is not a valid epoch-exhaustion retry) yields PRODUCER_FENCED (:252-253, isValidProducerId at :234-246).

2 · AddPartitionsToTxn, register partitions

Before producing to a partition, the client tells the coordinator about it. handleAddPartitionsToTransaction validates producerId/producerEpoch against the metadata (mismatches → INVALID_PRODUCER_ID_MAPPING / PRODUCER_FENCED), short-circuits with NONE if the partitions are already present, otherwise calls prepareAddPartitions and appends an ONGOING record (TransactionCoordinator.scala:416-468). The first AddPartitions sets txnStartTimestamp, which arms the timeout clock (TransactionMetadata.java:214-233).

There is also a server-internal verification path (handleVerifyPartitionsInTransaction, :361-414): when a partition leader receives a transactional produce it can ask the coordinator "is this partition really in this producer's open transaction?". This closes the hanging transaction hole where a delayed produce could land after a marker. The leader's side uses a VerificationGuard, a unique token created in maybeStartTransactionVerification and checked in analyzeAndValidateProducerState, to defeat the ABA race between a verified response and an abort marker (storage/.../UnifiedLog.java:1410-1440, see the long comment at :1414-1426). Verification is on by default (transaction.partition.verification.enable=true, TransactionLogConfig.java:52-54).

3 · Produce, transactional data batches

Produced batches carry the PID, epoch, sequence and the transactional flag. The leader runs the idempotence checks above and the verification check; the first transactional batch on a partition sets currentTxnFirstOffset in the producer state and starts a TxnMetadata in the ProducerStateManager.ongoingTxns map (storage/.../ProducerAppendInfo.java:234-243, ProducerStateManager.java:398). A non-transactional write while a transaction is open throws InvalidTxnStateException; the converse, a transactional write with no open transaction, is not an error, it is exactly how a transaction begins (sets currentTxnFirstOffset and adds a TxnMetadata) (ProducerAppendInfo.java:235-243).

4 · Consume-transform-produce: binding offsets (KIP-447)

For EOS pipelines the application commits its input consumer offsets inside the transaction. The producer client method sendOffsetsToTransaction(offsets, groupMetadata) (clients/.../TransactionManager.java:432-463) takes the full ConsumerGroupMetadata (group id, generation, member id), this is the KIP-447 improvement that lets the group coordinator fence offset commits by generation, so a single transactional producer can serve a partition without one-producer-per-input-partition. Under TV_1 it sends AddOffsetsToTxn (which registers the __consumer_offsets partition with the txn coordinator, exactly like AddPartitionsToTxn) then TxnOffsetCommit; under TV_2 it skips AddOffsetsToTxn and sends TxnOffsetCommit directly (:443-459). The committed offsets are written to __consumer_offsets by the group coordinator but stay invisible (LSO-gated) until the transaction's COMMIT marker lands on that partition.

5 · EndTxn, two-phase commit

handleEndTransaction → endTransaction implements the 2PC. There are two implementations selected by the client's TV: endTransactionWithTV1 (TransactionCoordinator.scala:543-710) and the TV_2 endTransaction (:758-1007). The shape is the same:

1. Phase 1 (prepare). Under the metadata lock, validate pid/epoch and current state, compute the target (PREPARE_COMMIT or PREPARE_ABORT) via prepareAbortOrCommit, and append that record to __transaction_state (:854-860, 1003-1004). Under TV_2, prepareAbortOrCommit bumps the epoch and, if the epoch is exhausted, rotates to nextProducerId (:818-841, TransactionMetadata.java:235-263).
2. Respond early. Once the prepare record is durable, the coordinator replies to the client immediately (returning the post-bump pid/epoch under TV_2, :975) and continues asynchronously, the client does not wait for markers.
3. Phase 2 (commit). addTxnMarkersToSend hands the transaction to the TransactionMarkerChannelManager, which writes COMMIT/ABORT markers to every involved partition leader. When all markers are acknowledged, the coordinator calls prepareComplete and appends COMPLETE_COMMIT/COMPLETE_ABORT (TransactionMarkerChannelManager.scala:276-315, TransactionMetadata.java:265-288).

The transaction marker (control record)

On the receiving partition leader, ProducerAppendInfo.appendEndTxnMarker validates the marker epoch (TV_2: strictly greater) and the coordinator epoch (a marker with an older coordinator epoch is a zombie and throws TransactionCoordinatorFencedException, ProducerAppendInfo.java:246-257). If the producer had an open transaction (currentTxnFirstOffset present), it emits a CompletedTxn recording (producerId, firstOffset, lastOffset=markerOffset, isAborted) (:259-290).

Marker fan-out: TransactionMarkerChannelManager

The channel manager is an InterBrokerSendThread named TxnMarkerSenderThread-<brokerId> (TransactionMarkerChannelManager.scala:169) with its own dedicated NetworkClient using the inter-broker listener. It owns several concurrent structures:

addTxnMarkersToSend records the pending completion, then addTxnMarkersToBrokerQueue groups the transaction's partitions by their leader (via metadataCache.getPartitionLeaderEndpoint) and enqueues a TxnMarkerEntry per destination (:317-458). The send thread's generateRequests drains per-broker queues into WriteTxnMarkersRequests, attaching a TransactionMarkerRequestCompletionHandler (:238-274). As each partition leader acknowledges, the handler removes that partition from the metadata's topicPartitions; when the set empties, maybeWriteTxnCompletion fires writeTxnCompletion → tryAppendToLog to persist the COMPLETE_* record (:339-383). A partition whose leader has been deleted is simply dropped from the set so completion can proceed (:421-445).

On the receiving broker, KafkaApis.handleWriteTxnMarkersRequest appends each marker to its partitions with an AtomicInteger numAppends counter, sending the response only after the last partition's append completes (core/src/main/scala/kafka/server/KafkaApis.scala:1729-1871).

Concurrency & threading model

The coordinator's locking discipline is documented at the top of TransactionStateManager (:60-68) and is strict:

• stateLock (a ReentrantReadWriteLock) guards the metadata cache and the loading/unloading partition sets. Loading and emigration take the write lock; request handling takes the read lock.
• Each TransactionMetadata has its own ReentrantLock for field mutation.
• Ordering rule: never acquire stateLock.readLock while holding a txnMetadata lock; never acquire a txnMetadata lock while holding stateLock.writeLock; never call ReplicaManager.appendRecords while holding a txnMetadata lock.

The append path holds the read lock across appendRecords deliberately (:772-813): otherwise an emigration+immigration between the coordinator-epoch check and the append could let a stale-epoch record replicate, corrupting the log. The post-append updateCacheCallback re-verifies the coordinator epoch before applying completeTransitionTo, mapping mismatches to NOT_COORDINATOR (:666-770).

The consumer side: read_committed, LSO & the aborted index

Computing the LSO on the leader

The leader tracks firstUnstableOffsetMetadata, the earliest offset belonging to an undecided or not-yet-replicated transaction. ProducerStateManager.firstUnstableOffset returns the min of the first ongoingTxns entry (undecided) and the first unreplicatedTxns entry (decided but its marker is above the HW) (ProducerStateManager.java:246-258). UnifiedLog.lastStableOffset is then min(highWatermark, firstUnstableOffset) (UnifiedLog.java:678-686). When a transaction completes, completeTxn moves it from ongoingTxns to unreplicatedTxns; it is only dropped (and the LSO allowed past it) once the high watermark advances beyond its marker, via onHighWatermarkUpdated (ProducerStateManager.java:545-554, 264-266).

The aborted-transaction index

When a transaction aborts, the segment's TransactionIndex gets an AbortedTxn entry. The on-disk record is fixed-size (no flexible fields), 4 × int64 = 32 bytes plus a 2-byte version prefix:

On a read_committed fetch the leader returns the relevant AbortedTransaction list (gathered by TransactionIndex.collectAbortedTxns) alongside the records, and the fetch is capped at the LSO (UnifiedLog.read passes isolation == TXN_COMMITTED, UnifiedLog.java:1659).

Client-side filtering

CompletedFetch.nextFetchedRecord performs the actual filtering (clients/.../CompletedFetch.java:187-244). It keeps a priority queue of aborted transactions ordered by first offset and a live abortedProducerIds set. For each batch with a producer id:

1. consumeAbortedTransactionsUpTo(batch.lastOffset()) moves every aborted txn whose first offset ≤ the batch into the active set (:356-364).
2. If the batch is an abort marker, the producer id is removed from the active set (the abort is now consumed).
3. If the batch's producer id is in the active set, the whole batch is skipped and nextFetchOffset advances past it (:215-223).
4. Control batches (the markers themselves) are never surfaced to the application (:234-240).

Producer-client state machine

TransactionManager mirrors the server with a client-side machine (clients/.../TransactionManager.java:157-195): UNINITIALIZED → INITIALIZING → READY → IN_TRANSACTION → {COMMITTING,ABORTING}_TRANSACTION → READY, plus PREPARED_TRANSACTION (2PC) and the error sinks ABORTABLE_ERROR / FATAL_ERROR. It maintains per-partition sequence counters (sequenceNumber(tp) → nextSequence(), :710-713), tracks in-flight batches for retry/reordering, and serializes coordinator interactions through a priority queue (FIND_COORDINATOR < INIT_PRODUCER_ID < ADD_PARTITIONS_OR_OFFSETS < END_TXN < EPOCH_BUMP, :201-213). Recoverable errors transition to ABORTABLE_ERROR so the next operation must be an abort; only an abortable retry of a bumped epoch reuses the machine. Under KIP-360, the client can bump its idempotent epoch and reset sequences to recover from UNKNOWN_PRODUCER_ID/OutOfOrderSequence without becoming fatal (requestIdempotentEpochBumpForPartition, bumpIdempotentEpochAndResetIdIfNeeded, :668-708). Details of batching and the sender loop are in The Producer Client.

Configuration reference

Sources: TransactionLogConfig.java, TransactionStateManagerConfig.java, TransactionVersion.java:35.

Failure modes, edge cases & recovery

Coordinator failover

When a broker becomes leader for a __transaction_state partition, TransactionCoordinator.onElection first clears any stale marker queues, then loadTransactionsForTxnTopicPartition replays the compacted partition into the cache (TransactionCoordinator.scala:477-487, TransactionStateManager.scala:538-597). Crucially, for any transaction loaded in PREPARE_COMMIT or PREPARE_ABORT, the new coordinator immediately re-drives phase 2, it calls prepareComplete and re-sends the markers (:562-588). This is what makes the prepare record the durable commit point: a crash between prepare and complete is transparently finished by the successor. The partition is removed from loadingPartitions before the markers are sent so the completion append is not blocked by a self-imposed COORDINATOR_LOAD_IN_PROGRESS (:579-582).

Log-append failures

appendTransactionToLog.updateCacheCallback maps storage errors to client-facing coordinator errors: NOT_ENOUGH_REPLICAS/REQUEST_TIMED_OUT → COORDINATOR_NOT_AVAILABLE (retriable); NOT_LEADER_OR_FOLLOWER/KAFKA_STORAGE_ERROR → NOT_COORDINATOR (TransactionStateManager.scala:680-697). On error it resets pendingState so the metadata is not stuck mid-transition (:734-767), unless the caller asked to retain it for retry. Completion-record appends that hit COORDINATOR_NOT_AVAILABLE are re-enqueued on txnLogAppendRetryQueue and retried by the sender thread (TransactionMarkerChannelManager.scala:354-383).

Epoch exhaustion & overflow

The coordinator never lets a client use epoch Short.MAX_VALUE (isEpochExhausted: producerEpoch >= Short.MAX_VALUE - 1, TransactionMetadata.java:64-66) so it always retains the ability to fence. When the epoch is about to overflow, the next commit/abort rotates to a pre-allocated nextProducerId and resets the epoch to 0 while keeping lastProducerEpoch = MAX-1 for client-visible continuity (prepareComplete, :265-288; validProducerEpoch overflow handling, :469-487). TV_2 retry detection (retryOnOverflow) recognizes a client replaying an EndTxn that overflowed: prevProducerId == producerId && producerEpoch == MAX-1 && metadata.producerEpoch == 0 (TransactionCoordinator.scala:791-792).

Hanging transactions & late writes

ProducerStateManager.hasLateTransaction flags a transaction whose oldest update is older than maxTransactionTimeoutMs + LATE_TRANSACTION_BUFFER_MS (5 min, ProducerStateManager.java:72, 130-133); this surfaces a metric/health signal so operators can detect a partition pinned by a stuck transaction (which would freeze its LSO and stall read_committed consumers). The verification path (above) is the primary prevention.

UNKNOWN_PRODUCER_ID

If retention or DeleteRecords removes a PID's last batch, the leader drops the ProducerStateEntry and a subsequent append with a non-zero sequence cannot be validated → UNKNOWN_PRODUCER_ID. The producer recovers by bumping its idempotent epoch and resetting sequences (KIP-360). Setting producer.id.expiration.ms ≥ delivery.timeout.ms avoids spurious expiration during retries.

Invariants & guarantees (summary)

• PID uniqueness: the controller's nextProducerId is strictly monotonic in the metadata log; blocks are never reissued (ProducerIdControlManager.java:111-120).
• Per-partition EOS-in-order: persisted batches for a fixed (pid, epoch) have contiguous sequence ranges from 0; retries are deduped (ProducerAppendInfo, ProducerStateEntry.findDuplicateBatch).
• Atomicity: a transaction's records become visible (pass the LSO) iff a COMMIT marker was written to every involved partition; the prepare record fixes the outcome before any marker (handleEndTransaction + failover replay).
• Zombie fencing: a write/marker with epoch below the leader's last-seen epoch (TV_2: not strictly greater) is rejected (checkProducerEpoch); a marker with an older coordinator epoch is rejected (checkCoordinatorEpoch).
• read_committed correctness: consumers never read past the LSO and filter every batch whose PID belongs to an aborted transaction beginning at/before it (CompletedFetch).

Interactions with other subsystems

• Record Format & Batches, the PID/epoch/baseSequence and transactional/control flags live in the v2 batch header; markers are control batches.
• The Log Storage Engine, ProducerStateManager, producer-state snapshots, the transaction index and LSO are part of UnifiedLog.
• Replication, ISR & High Watermark, the LSO can never exceed the HW; markers replicate like any record before influencing visibility.
• The Fetch Path & Replica Fetchers, read_committed fetches are capped at the LSO and carry the aborted-transaction list.
• The KRaft Controller, allocates PID blocks via AllocateProducerIds / ProducerIdsRecord.
• Group Coordination, TxnOffsetCommit writes transactional offsets to __consumer_offsets; KIP-447 binds them to the consumer group generation.
• The Producer Client / The Consumer Client, the client-side state machines that drive and observe transactions.
• Share Groups, share-group acknowledgements reuse the same PID/epoch machinery (KIP-932) for their internal state.

Design rationale & evolution

• KIP-98 introduced idempotent and transactional producers: PID + epoch + sequence, the __transaction_state log, two-phase commit and transaction markers, the foundation of this chapter.
• KIP-360 made the idempotent producer resilient: the client can bump its epoch and reset sequences to recover from UNKNOWN_PRODUCER_ID instead of failing fatally.
• KIP-447 changed sendOffsetsToTransaction to take ConsumerGroupMetadata so the group coordinator fences offset commits by generation, enabling one transactional producer per application instance rather than per input partition.
• KIP-890 (Transaction V2) bumps the producer epoch on every commit/abort so each transaction is uniquely identified by (pid, epoch). EndTxnResponse (v5+) returns the new pid/epoch; the client need not call InitProducerId except at startup. This closes the duplicate-across-transactions and hanging-transaction holes that required the server-side verification step on legacy versions. Enabled by the transaction.version feature flag (TransactionVersion.java:24-100).
• KIP-915 made the coordinator records flexible/tagged so future fields can be added without breaking downgrade (note in TransactionLogValue.json:21).

Gotchas & operational notes

• Partition count is permanent. Because partitionFor(transactionalId) = abs(hash) % transactionTopicPartitionCount (TransactionStateManager.scala:439) determines which coordinator owns an id, changing transaction.state.log.num.partitions after deploy reshuffles ownership and breaks fencing; the loader even throws KafkaException if it detects a change mid-load (:641-646).
• A stuck transaction freezes its partitions' LSO. Any partition with an undecided transaction holds back the LSO, stalling all read_committed consumers of that partition until the txn commits, aborts, or times out, watch the late-transaction metric and transaction.max.timeout.ms.
• Markers are written even for empty partitions sets carefully. Partitions whose leaders vanished are dropped from the set so completion can proceed; the marker for a deleted partition is skipped, not blocked (TransactionMarkerChannelManager.scala:432-445).
• 2PC transactions never time out. With transaction.two.phase.commit.enable=true a transaction's timeout becomes Integer.MAX_VALUE and the abort reaper skips it (TransactionMetadata.isDistributedTwoPhaseCommitTxn, :308-310), the external coordinator is responsible for resolution.
• Tune producer.id.expiration.ms vs. retries. If a PID's last record is retention-deleted while the producer is between sends, the next send hits UNKNOWN_PRODUCER_ID; aligning expiration with delivery.timeout.ms avoids surprise duplicates.