Digging into Kafka's Source 12: Idempotence and Transactions — How Exactly-Once Is Built, and Wrapping Up the Series

What this post covers

This is the twelfth and final post in the "Digging into Kafka's Source" series. In part 10 we saw the controller write metadata to a log and spread it, and in part 11 we saw a consumer group's offsets written as records to __consumer_offsets. This time we reach the last piece of the "what runs the cluster" group: the idempotent producer and transactions. In an application they are two lines of configuration.

enable.idempotence=true      # the default in 4.x
transactional.id=order-writer

What happens behind those two lines is the question of this post.

• If the producer retries, doesn't the same record get written twice? How does the broker recognise a duplicate?
• When you call commitTransaction(), who writes what, and where? What makes writes across several partitions become visible "all at once"?
• How does an isolation.level=read_committed consumer filter out records from aborted transactions? Does the broker filter them, or the consumer?
• What happens if two producers with the same transactional.id are running?

All line numbers in this post refer to this commit. How to get the same source and build the Docker image for the examples is in part 1.

In terms of the map from part 2, this post zooms into the transaction coordinator in region F (coordinators) and the producer's TransactionManager in region A (clients). Transactions don't stay in one region, though. Batches sent by the producer are checked in regions D and E (partitions and logs), the coordinator's markers are written to the same logs, and the consumer filters while reading those logs.

Blue boxes are the main characters of this post (producer, coordinator, the thread that sends markers, consumer); yellow boxes are where state is recorded. The letter in front is the region on the part 2 map.

The transaction code is not small. TransactionManager.java is 2,066 lines and TransactionCoordinator.scala is 1,145 lines, and the paths for transaction version 1 and 2 live side by side in the same files. This post picks only the main line for version 2, the default at this commit; what was left out is listed by name and location in "What this post leaves out" at the end.

Words to know first

Exactly-once: the promise that "a record is written to the log only once, and the reader also sees it as if it were processed only once". Networks lose responses and processes die halfway. Left alone, you get either "at least once" (retries can create duplicates) or "at most once" (no retries, so data can be lost). Kafka builds exactly-once in two layers: the idempotent producer, which prevents duplicates within one partition, and transactions, which make writes across several partitions visible all together or discarded all together.

Producer id · epoch · sequence: when an idempotent producer starts, it receives an integer from a broker — the producer id (PID). The epoch is the "generation number" of that producer id (a short that counts up from 0), and the sequence is a per-partition number that starts at 0 and grows by the number of records. These three go into the producerId, producerEpoch and baseSequence fields of the record batch header we saw in part 6. The broker decides about duplicates and gaps by looking at these three alone.

transactional.id: a name for a transactional producer chosen by the application. The producer id is a number handed out by a broker and can change when the process restarts, but the transactional.id is fixed by the application. The coordinator uses it as the key to remember "this producer's producer id and epoch, and the transaction that is open right now".

Transaction coordinator and __transaction_state: the broker-side component that manages transaction state. It isn't a separate server; there is one inside every broker, and the broker that leads a partition of the internal topic __transaction_state becomes the coordinator for the transactional.ids that map to that partition. It is the same shape as what the group coordinator does with __consumer_offsets in part 11.

Control records (transaction markers): special records written to each partition to mark that a transaction has ended. The isControl bit of the batch header is set, and the key is COMMIT or ABORT. A marker takes up an offset but is never delivered to the consumer application.

LSO (last stable offset): the boundary where "everything below this offset has been decided". Non-transactional records count as decided as soon as they are written; transactional records are decided once their marker is written. The LSO can't pass the high watermark (HW, the point every ISR member has received) from part 9. A read_committed consumer reads only up to the LSO.

Transaction version (transaction.version): one of the cluster feature flags. Like the metadata.version from part 2, the whole cluster uses a single value, and the transaction protocol changes depending on it.

    // Version 0 is the original transaction coordinator with no extra features enabled.
    TV_0(0, MetadataVersion.MINIMUM_VERSION, Map.of()),
    // Version 1 enables flexible transactional state records. (KIP-890)
    TV_1(1, MetadataVersion.IBP_4_0_IV2, Map.of()),
    // Version 2 enables epoch bump per transaction and optimizations. (KIP-890)
    TV_2(2, MetadataVersion.IBP_4_0_IV2, Map.of());

server-common/src/main/java/org/apache/kafka/server/common/TransactionVersion.java lines 26–31

The heart of version 2 (TV2) is two things: "bump the epoch once per transaction" and "the client no longer sends a separate request to add a partition to the transaction" (section 3). In this series' example cluster (formatted with metadata.version 4.4-IV2), TV2 is on — the first output of example 1 shows it. The 4.4.0 stable release has the same TransactionVersion.java (the file is identical).

1. The idempotent producer — so that retries don't create duplicates

What is the problem?

Suppose a leader receives a Produce request sent with acks=all, writes it to the log and finishes replicating it, but the connection drops while the response is on its way back. The producer sees a failure and sends the same batch again (the Sender retries we saw in part 5). If the broker remembered nothing, the same records would be written twice. An idempotent producer (idempotent: doing it many times has the same effect as doing it once) makes these retries harmless.

In 4.x you get an idempotent producer without turning anything on. The default is in ProducerConfig.

                                .define(ENABLE_IDEMPOTENCE_CONFIG,
                                        Type.BOOLEAN,
                                        true,
                                        Importance.LOW,
                                        ENABLE_IDEMPOTENCE_DOC)

clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java lines 574–578

The default for acks is also "all" (lines 430–435 of the same file). When idempotence is on only by default and the user sets acks=1, it is turned off with just one INFO log line; but if the user explicitly wrote enable.idempotence=true and acks isn't all, a ConfigException is thrown (lines 663–669). A max.in.flight.requests.per.connection greater than 5 is also rejected (lines 672–676). Where that number 5 comes from, we'll see on the broker side shortly.

Client: the sequence is attached at drain time, not in send()

The producer id is fetched when the Sender thread first runs: TransactionManager.bumpIdempotentEpochAndResetIdIfNeeded() queues an InitProducerId request (with transactionalId set to null) (TransactionManager.java lines 691–705). Surprisingly, the sequence number is not attached in send(). It is attached in RecordAccumulator.drain() from part 5 — the moment a batch is taken off the queue and put into a request.

                boolean isTransactional = transactionManager != null && transactionManager.isTransactional();
                ProducerIdAndEpoch producerIdAndEpoch =
                    transactionManager != null ? transactionManager.producerIdAndEpoch() : null;
                if (producerIdAndEpoch != null && !batch.hasSequence()) {
                    // If the producer id/epoch of the partition do not match the latest one
                    // of the producer, we update it and reset the sequence. This should be
                    // only done when all its in-flight batches have completed. This is guarantee
                    // in `shouldStopDrainBatchesForPartition`.
                    transactionManager.maybeUpdateProducerIdAndEpoch(batch.topicPartition);

                    // If the batch already has an assigned sequence, then we should not change the producer id and
                    // sequence number, since this may introduce duplicates. In particular, the previous attempt
                    // may actually have been accepted, and if we change the producer id and sequence here, this
                    // attempt will also be accepted, causing a duplicate.
                    //
                    // Additionally, we update the next sequence number bound for the partition, and also have
                    // the transaction manager track the batch so as to ensure that sequence ordering is maintained
                    // even if we receive out of order responses.
                    batch.setProducerState(producerIdAndEpoch, transactionManager.sequenceNumber(batch.topicPartition), isTransactional);
                    transactionManager.incrementSequenceNumber(batch.topicPartition, batch.recordCount);
                    log.debug("Assigned producerId {} and producerEpoch {} to batch with base sequence " +
                            "{} being sent to partition {}", producerIdAndEpoch.producerId,
                        producerIdAndEpoch.epoch, batch.baseSequence(), tp);

                    transactionManager.addInFlightBatch(batch);

clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java lines 1024–1048

The key is the !batch.hasSequence() condition. A sequence is attached only once, when the batch goes out for the first time; a retried batch goes again with the producer id, epoch and sequence it received the first time. The comment gives the reason: the earlier attempt may actually have been accepted, so if the numbers changed, the broker couldn't recognise the duplicate. The next batch's sequence skips ahead by the record count via incrementSequenceNumber(). If a batch holds 3 records it uses sequences 0–2, and the next batch starts at 3.

Broker: it remembers the last 5 batches

The broker keeps a ProducerStateManager per partition and remembers one ProducerStateEntry per producer id. That entry holds the metadata of recent batches (first and last sequence, offsets) — up to 5 of them.

public class ProducerStateEntry {
    public static final int NUM_BATCHES_TO_RETAIN = 5;

storage/src/main/java/org/apache/kafka/storage/internals/log/ProducerStateEntry.java lines 34–35

When 5 are stored, the oldest is dropped (addBatchMetadata(), lines 101–104). The client-side upper limit of 5 for max.in.flight.requests.per.connection is matched to exactly this number, as the comment on line 295 of ProducerConfig.java says ("The value 5 is aligned with ProducerStateEntry#NUM_BATCHES_TO_RETAIN"). If at most 5 requests per connection can be waiting for a response, a batch that comes back as a retry is always among the 5 the broker remembers.

In part 6, after UnifiedLog.append() assigned offsets, we only said "it checks the producer state" and moved on. This is that place.

        for (MutableRecordBatch batch : records.batches()) {
            if (batch.hasProducerId()) {
                // if this is a client produce request, there will be up to 5 batches which could have been duplicated.
                // If we find a duplicate, we return the metadata of the appended batch to the client.
                if (origin == AppendOrigin.CLIENT) {
                    Optional<ProducerStateEntry> maybeLastEntry = producerStateManager.lastEntry(batch.producerId());

                    Optional<BatchMetadata> duplicateBatch = maybeLastEntry.flatMap(e -> e.findDuplicateBatch(batch));
                    if (duplicateBatch.isPresent()) {
                        return new AnalyzeAndValidateProducerStateResult(updatedProducers, completedTxns, duplicateBatch);
                    }
                }

storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java lines 1411–1422

findDuplicateBatch() asks: "is there a remembered batch with the same epoch whose first and last sequence are exactly the same?" (ProducerStateEntry.java lines 128–137). If so, append() writes nothing to the log and returns, in the response, the offset at which that batch was first written (UnifiedLog.java lines 1261–1268). From the producer's point of view, it gets the same answer as the first time: "success, offset 0".

If it isn't a duplicate, ProducerAppendInfo.checkSequence() checks the order.

        if (producerEpoch != updatedEntry.producerEpoch()) {
            if (appendFirstSeq != 0) {
                if (updatedEntry.producerEpoch() != RecordBatch.NO_PRODUCER_EPOCH) {
                    throw new OutOfOrderSequenceException("Invalid sequence number for new epoch of producer " + producerId +
                            " at offset " + offset + " in partition " + topicPartition + ": " + producerEpoch + " (request epoch), "
                            + appendFirstSeq + " (seq. number), " + updatedEntry.producerEpoch() + " (current producer epoch)");
                }
            }
        } else {
            int currentLastSeq;
            if (!updatedEntry.isEmpty())
                currentLastSeq = updatedEntry.lastSeq();
            else if (producerEpoch == currentEntry.producerEpoch())
                currentLastSeq = currentEntry.lastSeq();
            else
                currentLastSeq = RecordBatch.NO_SEQUENCE;

            // If there is no current producer epoch (possibly because all producer records have been deleted due to
            // retention or the DeleteRecords API) accept writes with any sequence number
            if (!(currentEntry.producerEpoch() == RecordBatch.NO_PRODUCER_EPOCH || inSequence(currentLastSeq, appendFirstSeq))) {
                throw new OutOfOrderSequenceException("Out of order sequence number for producer " + producerId + " at " +
                        "offset " + offset + " in partition " + topicPartition + ": " + appendFirstSeq +
                        " (incoming seq. number), " + currentLastSeq + " (current end sequence number)");
            }
        }
    }

    private boolean inSequence(int lastSeq, int nextSeq) {
        return nextSeq == lastSeq + 1L || (nextSeq == 0 && lastSeq == Integer.MAX_VALUE);
    }

storage/src/main/java/org/apache/kafka/storage/internals/log/ProducerAppendInfo.java lines 181–210

There are two rules. If the epoch is new, the sequence must start at 0; if the epoch is the same, it must start at the previous batch's last sequence + 1 (after Integer.MAX_VALUE it wraps to 0). Before this, checkProducerEpoch() rejects an epoch lower than the remembered one with InvalidProducerEpochException (lines 122–159). Put together:

The duplicate check is done in UnifiedLog.analyzeAndValidateProducerState(), the epoch and order checks in ProducerAppendInfo. The special rule for when there is no producer state at all (for example, everything was deleted by retention) is left out of the diagram.

Example 2: the same batch twice, and one batch that skips sequences

This is easier to believe when you see it, so let's do what an idempotent producer does by hand, without KafkaProducer. As in the raw-socket example in part 3, we build requests with the generated …Data classes and send them. This example was run with Kafka 4.5.0-SNAPSHOT built as-is from the reference commit (ab1e816411), in a Docker container (Eclipse Temurin JDK 25). The same goes for every example below.

// Example written for this post (not Kafka source code)
// ...
            // 2) sequences 0..2 (3 records) — the first batch sent
            produce(topic, 0, "first send", "a", "b", "c");
            // the exact same batch again — the producer retried because the response was lost
            produce(topic, 0, "same batch again", "a", "b", "c");
            // the next sequence should be 3, but this starts at 5 — a batch in between went missing
            produce(topic, 5, "skip seq 3..4", "x");
            // the correct next sequence
            produce(topic, 3, "next seq", "d", "e", "f");
            // send four more batches so the first one falls out of the broker's memory (the last 5 batches)
            for (int seq = 6; seq <= 15; seq += 3) produce(topic, seq, "more", "g", "h", "i");
            // now what if we resend the first batch (sequences 0..2)?
            produce(topic, 0, "oldest of the 6 batches, again", "a", "b", "c");
// ...
    static void produce(String topic, int baseSequence, String label, String... values) throws Exception {
        // ...
        MemoryRecords records = MemoryRecords.withIdempotentRecords(Compression.NONE, producerId, epoch, baseSequence, recs);
        // ... put this batch in a Produce v12 request (acks=-1), send it, and print the response's error and baseOffset

Shortened from ex/u12-idempotence/RawIdempotent.java. The beginning is code that sends an InitProducerIdRequestData (transactionalId=null) to get a producer id. Here is the output.

=== raw InitProducerId + Produce
InitProducerId: COORDINATOR_LOAD_IN_PROGRESS, retrying in 100 ms
InitProducerId: error=NONE producerId=0 epoch=0
seq  0.. 2 (first send)                       -> error=NONE baseOffset=0
seq  0.. 2 (same batch again)                 -> error=NONE baseOffset=0
seq  5.. 5 (skip seq 3..4)                    -> error=OUT_OF_ORDER_SEQUENCE_NUMBER baseOffset=-1
seq  3.. 5 (next seq)                         -> error=NONE baseOffset=3
seq  6.. 8 (more)                             -> error=NONE baseOffset=6
seq  9..11 (more)                             -> error=NONE baseOffset=9
seq 12..14 (more)                             -> error=NONE baseOffset=12
seq 15..17 (more)                             -> error=NONE baseOffset=15
seq  0.. 2 (oldest of the 6 batches, again)   -> error=OUT_OF_ORDER_SEQUENCE_NUMBER baseOffset=-1
=== kafka-dump-log: u12-idem-0 batches
baseOffset: 0 lastOffset: 2 count: 3 baseSequence: 0 lastSequence: 2 producerId: 0 producerEpoch: 0 magic: 2
baseOffset: 3 lastOffset: 5 count: 3 baseSequence: 3 lastSequence: 5 producerId: 0 producerEpoch: 0 magic: 2
baseOffset: 6 lastOffset: 8 count: 3 baseSequence: 6 lastSequence: 8 producerId: 0 producerEpoch: 0 magic: 2
baseOffset: 9 lastOffset: 11 count: 3 baseSequence: 9 lastSequence: 11 producerId: 0 producerEpoch: 0 magic: 2
baseOffset: 12 lastOffset: 14 count: 3 baseSequence: 12 lastSequence: 14 producerId: 0 producerEpoch: 0 magic: 2
baseOffset: 15 lastOffset: 17 count: 3 baseSequence: 15 lastSequence: 17 producerId: 0 producerEpoch: 0 magic: 2

(The kafka-dump-log.sh output was trimmed with sed to keep only the fields we need.)

• The second send gets baseOffset=0 with no error. Yet on disk there is only one batch with sequences 0–2. The broker recognised the duplicate and returned the original offset.
• The batch that skipped to sequence 5 is rejected with OUT_OF_ORDER_SEQUENCE_NUMBER and isn't on disk. The expected next number was 3.
• The last line shows the limit of remembering 5. The batch with sequences 0–2 was pushed out of memory after 5 more batches were written. So the broker can't recognise it as a "duplicate" and rejects it in the order check (the next number after last sequence 17 is 18). A real KafkaProducer keeps in-flight requests at 5 or fewer, so it never hits this case.

Easy to get wrong: the controller hands out producer ids 1000 at a time

The COORDINATOR_LOAD_IN_PROGRESS on the first line of the example is a response I actually got on the first run. On that first run the code ignored this response and sent batches with producer id -1; the broker accepted them as non-idempotent batches, and the same batch was written twice (batches with producerId: -1 skip the checks above). The code above is what I got after adding a retry loop.

The reason lies in how producer ids are handed out. The broker's RPCProducerIdManager.generateProducerId() takes one number from the block of ids it holds. If the block is empty, it asks the controller for the next block and throws COORDINATOR_LOAD_IN_PROGRESS in the meantime (RPCProducerIdManager.java lines 80–109). On the controller side, ProducerIdControlManager.generateNextProducerId() creates a block of ProducerIdsBlock.PRODUCER_ID_BLOCK_SIZE (1000) ids and records the first id of the next block in the metadata log as a ProducerIdsRecord (ProducerIdControlManager.java lines 84–101). It's the same "the controller decides, writes it to the log, and spreads it" path from part 10, so right after a broker starts, its first block may not have arrived yet. KafkaProducer treats this error as retriable, so applications never see it.

2. The stage for transactions — the coordinator and __transaction_state

An idempotent producer prevents duplicates only within one partition. If processing one order writes to both an orders topic and a payments topic, and you also want to move the read position (the consumer offset) along with them, these three must be "all visible or all invisible". The one that makes and remembers that decision is the transaction coordinator.

Who is my coordinator?

A partition of __transaction_state is chosen from the hash of the transactional.id, and that partition's leader is the coordinator.

  def partitionFor(transactionalId: String): Int = Utils.abs(transactionalId.hashCode) % transactionTopicPartitionCount

core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala line 451

Reading the Scala — def f(x: String): Int = expr is a method returning Int, and the expression after = is the return value (no return needed).

The number of partitions (transaction.state.log.num.partitions) defaults to 50, the replication factor to 3, and min.insync.replicas to 2 (TransactionLogConfig.java lines 32, 40, 45). This series' example helper has only one broker, so it lowers the replication factor and min ISR to 1. The client finds the coordinator by asking any broker with a FindCoordinator request (key type TRANSACTION); the broker picks the partition with txnCoordinator.partitionFor(key) and returns its leader (KafkaApis.scala line 1282). It's the same way the group coordinator picks a __consumer_offsets partition from the group.id.

What the coordinator remembers: TransactionMetadata and eight states

The coordinator keeps one TransactionMetadata object in memory per transactional.id. It holds the producer id, epoch, previous epoch, timeout, state, and the list of partitions in this transaction (TransactionMetadata.java lines 37–60). Every time the state changes, it writes the contents as a record to __transaction_state, and only after that write has been replicated does it change the in-memory state (the comment in completeTransitionTo(), lines 357–358). If the broker acting as coordinator dies, the new leader reads this partition from the beginning and rebuilds the same state — the same "the log is the state" design as the group coordinator in part 11.

The states are gathered in a single Java enum, and which state can follow which is written as one table.

    public static final Map<TransactionState, Set<TransactionState>> VALID_PREVIOUS_STATES = Map.of(
        EMPTY, Set.of(EMPTY, COMPLETE_COMMIT, COMPLETE_ABORT),
        ONGOING, Set.of(ONGOING, EMPTY, COMPLETE_COMMIT, COMPLETE_ABORT),
        PREPARE_COMMIT, Set.of(ONGOING, EMPTY, COMPLETE_COMMIT, COMPLETE_ABORT),
        PREPARE_ABORT, Set.of(ONGOING, PREPARE_EPOCH_FENCE, EMPTY, COMPLETE_COMMIT, COMPLETE_ABORT),
        COMPLETE_COMMIT, Set.of(PREPARE_COMMIT),
        COMPLETE_ABORT, Set.of(PREPARE_ABORT),
        DEAD, Set.of(EMPTY, COMPLETE_ABORT, COMPLETE_COMMIT),
        PREPARE_EPOCH_FENCE, Set.of(ONGOING)
    );

transaction-coordinator/src/main/java/org/apache/kafka/coordinator/transaction/TransactionState.java lines 101–110

The key is "the state you want to move to" and the value is "the states you may come from". As a diagram:

Blue boxes are the states where "the decision is made and markers are being written"; green boxes are finished states. The TV2-only arrows for "committing or aborting an empty transaction" (Empty/Complete → Prepare*), as well as Empty → Empty and Empty → Dead, are left out of the diagram.*

The most important boundary is the Prepare states. The moment PrepareCommit is written to __transaction_state, the decision "this transaction will be committed" is durably recorded. Even if the coordinator dies after that, the new coordinator meets PrepareCommit while reading the log and continues writing the markers (TransactionStateManager.loadTransactionsForTxnTopicPartition(), from line 550). It plays the same role as the "prepare" phase of a two-phase commit (2PC).

Easy to get wrong: where it differs from 4.4.0

The PREPARE_COMMIT line was Set.of(ONGOING) in 4.4.0. In trunk, calling commitTransaction() with no partitions (an empty transaction) now bumps the epoch and writes PrepareCommit to the log, just like an abort does. According to the comment below the state table in TransactionCoordinator.scala, the change returns a retriable error instead of a fatal one to the client at the moment when a coordinator has lost leadership but hasn't yet applied that fact (KAFKA-21181). There's almost no visible difference for applications, but if you read the 4.4.0 source alongside, it's worth knowing this line differs.

3. The life of one transaction — from initTransactions() to commitTransaction()

In application code it looks like this:

producer.initTransactions();
producer.beginTransaction();
producer.send(new ProducerRecord<>("u12-txn", "c1"));
// ...
producer.commitTransaction();

Drawing the requests exchanged during these five lines, with TV2, gives the following.

The coordinator and the partition leader may be the same broker or different brokers. Steps 5 and 12 are requests one broker sends to another.

Let's follow the numbers into the source.

1–3: InitProducerId — take your seat by bumping the epoch

initTransactions() goes through TransactionManager.initializeTransactions() and sends an InitProducerId request (this time with transactionalId filled in) (lines 321–356). The client's TransactionManager puts requests into a priority queue, and the order is fixed as FIND_COORDINATOR → INIT_PRODUCER_ID → ADD_PARTITIONS_OR_OFFSETS → END_TXN (followed by one more, EPOCH_BUMP, for bumping the epoch while ending a transaction; enum Priority, lines 201–213). If you don't know the coordinator, finding it comes first; if you have no producer id, getting one comes first.

The coordinator's handleInitProducerId() gets a new producer id and creates a TransactionMetadata if it's a transactional.id it has never seen (lines 152–167), and then, new or existing, prepares an Empty state with the epoch bumped by one via prepareInitProducerIdTransit() and writes it to the log (the first time, epoch -1 becomes 0) (TransactionCoordinator.scala lines 118–227). This behaviour — a producer restarted with the same transactional.id gets a new epoch — is the fencing of section 6.

4–8: adding partitions to the transaction is now the broker's job

Up to TV1, the client sent an AddPartitionsToTxn request directly to the coordinator before writing to a new partition. In TV2 it doesn't send this request. Look at maybeAddPartition(), called from send(), and you'll see the branch.

    public synchronized void maybeAddPartition(TopicPartition topicPartition) {
        maybeFailWithError();
        throwIfPendingState(TransactionOperation.SEND);

        if (isTransactional()) {
            if (!hasProducerId()) {
                throw new IllegalStateException("Cannot add partition " + topicPartition +
                    " to transaction before completing a call to initTransactions");
            } else if (currentState != State.IN_TRANSACTION) {
                throw new IllegalStateException("Cannot add partition " + topicPartition +
                    " to transaction while in state  " + currentState);
            } else if (isTransactionV2Enabled()) {
                txnPartitionMap.getOrCreate(topicPartition);
                partitionsInTransaction.add(topicPartition);
                transactionStarted = true;
            } else if (transactionContainsPartition(topicPartition) || isPartitionPendingAdd(topicPartition)) {
                return;
            } else {
                log.debug("Begin adding new partition {} to transaction", topicPartition);
                txnPartitionMap.getOrCreate(topicPartition);
                newPartitionsInTransaction.add(topicPartition);
            }
        }
    }

clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java lines 465–488

The TV2 branch immediately records the partition as "in the transaction" (partitionsInTransaction) and is done. The TV1 branch puts it into newPartitionsInTransaction and later sends a separate AddPartitionsToTxn request. Whether TV2 is on is decided by whether the transaction.version feature flag, carried in the broker's ApiVersions response, is 2 or higher (maybeUpdateTransactionV2Enabled(), lines 520–532).

So how does the coordinator learn about the partition? The partition leader that receives the batch tells it on the producer's behalf. KafkaApis.handleProduceRequest() decides what to do based on the version of the Produce request.

    public static TransactionSupportedOperation produceRequestVersionToTransactionSupportedOperation(short version) {
        if (version > 11) {
            return ADD_PARTITION;
        } else if (version > 10) {
            return GENERIC_ERROR_SUPPORTED;
        } else {
            return DEFAULT_ERROR;
        }
    }

server/src/main/java/org/apache/kafka/server/transaction/AddPartitionsToTxnManager.java lines 98–106

Produce v12 or later means ADD_PARTITION — "don't just verify, add it too". When there are transactional batches, ReplicaManager.handleProduceAppend() calls maybeSendPartitionsToTransactionCoordinator() before writing to the log (ReplicaManager.scala lines 1000–1059), and if that producer has no open transaction on this partition yet, AddPartitionsToTxnManager.addOrVerifyTransaction() sends an AddPartitionsToTxn request to the coordinator. With setVerifyOnly(!transactionSupportedOperation.supportsEpochBump) it asks to "add" for TV2 and to "verify only" for older clients (line 259). A separate thread inside the broker sends this request, and its name is set in the constructor.

        super("AddPartitionsToTxnSenderThread-" + config.brokerId(), client, config.requestTimeoutMs(), time);

server/src/main/java/org/apache/kafka/server/transaction/AddPartitionsToTxnManager.java line 226

On the coordinator side, handleAddPartitionsToTransaction() checks that the producer id and epoch match (a different producer id gives INVALID_PRODUCER_ID_MAPPING, a different epoch PRODUCER_FENCED) and writes an Ongoing state including the partition to the log (TransactionCoordinator.scala lines 417–470). When the answer comes back, the leader finally writes the batch to the log.

Why the change? The reason you can confirm in the source is in the long comment in UnifiedLog.analyzeAndValidateProducerState() (lines 1425–1440). It guarantees that "transactional records are never written to the log without the coordinator knowing there is an open transaction on that partition". If the client forgets or delays AddPartitionsToTxn, the coordinator doesn't know to send a marker to that partition, an unfinished transaction is left there, and the LSO stops forever. Having the leader that receives the write check and add it itself closes this hole (the VerificationGuard mentioned in that comment is a device that also covers the case where the verification response and an abort marker cross each other).

9–11: EndTxn — write Prepare, and answer right away

commitTransaction() sends an EndTxn request through beginCommit() → beginCompletingTransaction() (lines 381–430). The coordinator's endTransaction() prepares PrepareCommit (or PrepareAbort) if the state is Ongoing, and under TV2 this is where it bumps the epoch.

        TransitionData data = new TransitionData(newState);
        if (clientTransactionVersion.supportsEpochBump()) {
            if (producerEpoch == Short.MAX_VALUE && newState == TransactionState.PREPARE_ABORT) {
                // If we're already in a broken state, we let the abort go through without
                // epoch overflow, so that we can recover and continue.
                LOGGER.error("Aborting producer {} {} with epoch equal to Short.MaxValue, this must not happen unless there is a bug", transactionalId, producerId);
            } else {
                // We already ensured that we do not overflow here. MAX_SHORT is the highest possible value.
                data.producerEpoch = (short) (producerEpoch + 1);
            }
            data.lastProducerEpoch = producerEpoch;
        } else {
            data.producerEpoch = producerEpoch;
            data.lastProducerEpoch = lastProducerEpoch;
        }

transaction-coordinator/src/main/java/org/apache/kafka/coordinator/transaction/TransactionMetadata.java lines 240–254

It writes this state to __transaction_state with appendTransactionToLog(), and when the write is done, the callback does this:

                case Right((txnMetadata, newPreSendMetadata)) =>
                  // we can respond to the client immediately and continue to write the txn markers if
                  // the log append was successful
                  responseCallback(Errors.NONE, newPreSendMetadata.producerId, newPreSendMetadata.producerEpoch)

                  txnMarkerChannelManager.addTxnMarkersToSend(coordinatorEpoch, txnMarkerResult, txnMetadata, newPreSendMetadata)

core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala lines 1002–1007

Reading the Scala — Right(...) and Left(...) are values of Either, an "one of two" type. This codebase puts errors in Left and success values in Right. case Right((a, b)) => is pattern matching that takes the success value (a tuple) apart and names its parts a and b.

It answers the client before all markers are written. Once PrepareCommit is in the log the decision can't be undone, so the coordinator just takes responsibility for finishing the rest. That's why, right after commitTransaction() returns, the markers may not be written yet — which is why example 1 waits half a second before its last output. The response carries the bumped epoch, and the client's EndTxnHandler takes that value as its own producer id and epoch and starts counting sequences from 0 again (TransactionManager.java lines 1817–1824).

12–14: sending markers and CompleteCommit

addTxnMarkersToSend() groups the partitions in the transaction by leader broker and puts them in queues (addTxnMarkersToBrokerQueue(), TransactionMarkerChannelManager.scala lines 339–412). The one that drains the queues is TransactionMarkerChannelManager itself, which extends InterBrokerSendThread; its thread name is "TxnMarkerSenderThread-" + config.brokerId (line 123). generateRequests() builds one WriteTxnMarkers request per broker (lines 192–228). When every partition has answered, writeTxnCompletion() writes CompleteCommit to the log (lines 230–269).

The leader that receives this WriteTxnMarkers writes the markers to the partition, and the core of it is an ordinary write.

            controlRecords += replicaManager.topicIdPartition(partition) -> MemoryRecords.withEndTransactionMarker(
              producerId,
              marker.producerEpoch,
              new EndTransactionMarker(controlRecordType, marker.coordinatorEpoch)
            )

core/src/main/scala/kafka/server/KafkaApis.scala lines 1867–1871

Reading the Scala — map += key -> value adds one entry to a mutable map. a -> b is a short way to write the tuple (a, b).

It builds a batch holding one marker record and hands it to ReplicaManager.appendRecords() from part 6 with origin = AppendOrigin.COORDINATOR and requiredAcks = -1 (lines 1876–1889). A marker, too, only succeeds once it's replicated. The exception is __consumer_offsets partitions. Consumer offsets added to the transaction with sendOffsetsToTransaction() are held by the group coordinator, so markers for that partition go to groupCoordinator.completeTransaction() (lines 1840–1862). Thanks to this, the "read position" is also committed or discarded together with the output records.

Easy to get wrong: sendOffsetsToTransaction() under TV2

Under TV1, adding offsets first sent AddOffsetsToTxn to the transaction coordinator and then TxnOffsetCommit to the group coordinator. Under TV2 the first request is skipped and only TxnOffsetCommit is sent (the comment "In transaction V2, the client will skip sending AddOffsetsToTxn" around lines 443–449 of TransactionManager.java). Adding the __consumer_offsets partition to the transaction is done by the receiving broker, the same way as for Produce (txnOffsetCommitRequestVersionToTransactionSupportedOperation(), AddPartitionsToTxnManager.java lines 108–116). No need to be surprised if you search the client log for AddOffsetsToTxn and find nothing.

4. Commit markers — looking at the disk

Example 1: commit, abort, commit

Let's send three transactions and open the disk. A (c1–c3) commits, B (a1–a3) aborts, C (c4–c5) commits. For B, after sending and calling flush() but before aborting, we ask for the end offset at both isolation levels (used in section 5).

// Example written for this post (not Kafka source code)
// ...
            producer.initTransactions();   // InitProducerId — receive a producer id and epoch

            // transaction A: commit
            producer.beginTransaction();
            for (String v : List.of("c1", "c2", "c3")) producer.send(new ProducerRecord<>(TOPIC, v));
            producer.commitTransaction();
            System.out.println("txn A (c1..c3): committed");

            // transaction B: send and flush, then ask for the end offset at both isolation levels before aborting
            producer.beginTransaction();
            for (String v : List.of("a1", "a2", "a3")) producer.send(new ProducerRecord<>(TOPIC, v));
            producer.flush();
            System.out.println("txn B (a1..a3): sent and flushed, still open");
            printEnd(admin, "  while B is open");
            producer.abortTransaction();
            System.out.println("txn B: aborted");
// ...

Shortened from ex/u12-transactions/Txn.java (the only extra producer settings are transactional.id=u12-txn and client.id, and the topic u12-txn has 1 partition). run.sh starts the server, runs this program, and then prints the disk and tool outputs in turn. First the feature flags and the program's output:

=== finalized features (transaction.version)
Feature: metadata.version                          SupportedMinVersion: 3.3-IV3          SupportedMaxVersion: 4.4-IV2          FinalizedVersionLevel: 4.4-IV2          Epoch: 72
Feature: transaction.version                       SupportedMinVersion: 0                SupportedMaxVersion: 2                FinalizedVersionLevel: 2                Epoch: 72
=== producer and consumers
txn A (c1..c3): committed
txn B (a1..a3): sent and flushed, still open
  while B is open: latest offset read_uncommitted=7 (high watermark), read_committed=4 (LSO)
txn B: aborted
txn C (c4..c5): committed
  after all three: latest offset read_uncommitted=11 (high watermark), read_committed=11 (LSO)
read_uncommitted consumer got 8 records: c1@0 c2@1 c3@2 a1@4 a2@5 a3@6 c4@8 c5@9 (position 11)
read_committed   consumer got 5 records: c1@0 c2@1 c3@2 c4@8 c5@9 (position 11)
=== broker threads that only transactions use
"AddPartitionsToTxnSenderThread-1"
"TxnMarkerSenderThread-1"

transaction.version is finalized at 2 (the Epoch value differs from run to run). The broker's jstack shows the two threads from section 3, carrying node number 1. Next, the partition folder and kafka-dump-log.sh:

=== files of u12-txn-0
00000000000000000000.index
00000000000000000000.log
00000000000000000000.timeindex
00000000000000000000.txnindex
=== kafka-dump-log: u12-txn-0 batches
baseOffset: 0 lastOffset: 0 baseSequence: 0 producerId: 0 producerEpoch: 0 isTransactional: true isControl: false magic: 2
baseOffset: 1 lastOffset: 2 baseSequence: 1 producerId: 0 producerEpoch: 0 isTransactional: true isControl: false magic: 2
baseOffset: 3 lastOffset: 3 baseSequence: -1 producerId: 0 producerEpoch: 1 isTransactional: true isControl: true magic: 2
 offset: 3 keySize: 4 valueSize: 6 sequence: -1 headerKeys: [] endTxnMarker: COMMIT coordinatorEpoch: 0
baseOffset: 4 lastOffset: 6 baseSequence: 0 producerId: 0 producerEpoch: 1 isTransactional: true isControl: false magic: 2
baseOffset: 7 lastOffset: 7 baseSequence: -1 producerId: 0 producerEpoch: 2 isTransactional: true isControl: true magic: 2
 offset: 7 keySize: 4 valueSize: 6 sequence: -1 headerKeys: [] endTxnMarker: ABORT coordinatorEpoch: 0
baseOffset: 8 lastOffset: 9 baseSequence: 0 producerId: 0 producerEpoch: 2 isTransactional: true isControl: false magic: 2
baseOffset: 10 lastOffset: 10 baseSequence: -1 producerId: 0 producerEpoch: 3 isTransactional: true isControl: true magic: 2
 offset: 10 keySize: 4 valueSize: 6 sequence: -1 headerKeys: [] endTxnMarker: COMMIT coordinatorEpoch: 0
=== kafka-dump-log: .txnindex
Dumping /tmp/kafka/1/data/u12-txn-0/00000000000000000000.txnindex
version: 0 producerId: 0 firstOffset: 4 lastOffset: 7 lastStableOffset: 8

(Some fields were removed from the batch lines with sed, and data-record lines were dropped with grep. Transaction A was split into two batches (0 and 1–2) because in this run the first record went out first. In another run 0–2 was a single batch — batch boundaries can differ between runs.)

Green is committed data, red is aborted data, grey is control records (markers). Markers also take up one offset each.

There are three things to read here.

• The epoch goes up by one per transaction. Data has 0, 1, 2; markers have 1, 2, 3. Because the TV2 coordinator bumps the epoch at Prepare* (section 3) and writes the marker with that bumped epoch, a marker's epoch is always 1 higher than the data of the same transaction. The next transaction starts with that epoch, and the sequence starts again at 0 (baseSequence: 0).
• A marker is a batch with baseSequence: -1 and isControl: true. It holds one record with a 4-byte key and 6-byte value, and the value contains the coordinator epoch.
• .txnindex holds only aborted transactions. The two commit markers created no entries, because LogSegment.updateTxnIndex() appends only when completedTxn.isAborted() (lines 348–354). The shape of an entry is generated from the JSON schema AbortedTxn.json — four fields ProducerId, FirstOffset, LastOffset, LastStableOffset, all int64 (clients/src/main/resources/common/message/AbortedTxn.json lines 22–33). The TransactionIndex we only saw by name in part 7 is this file.

The epoch check on the receiving side of a marker

The broker also checks the epoch with ProducerAppendInfo when it writes a marker. Under TV2 the rule is stricter than for data.

    private void checkProducerEpoch(short producerEpoch, long offset, short transactionVersion) {
        short current = updatedEntry.producerEpoch();
        boolean invalidEpoch = (transactionVersion >= 2) ? (producerEpoch <= current) : (producerEpoch < current);

storage/src/main/java/org/apache/kafka/storage/internals/log/ProducerAppendInfo.java lines 122–124

Data batches pass transactionVersion as TV_UNKNOWN (-1), so the rule is "reject if lower" (lines 97–98); TV2 markers are "reject if equal or lower". The "marker epoch = data epoch + 1" from the example just above is the premise of this check. The method comment (lines 104–121) explains that this prevents an old marker arriving late from wrongly ending the next transaction. There is an exception on lines 127–144, though: if a marker with the same epoch comes again as a retry and the transaction has already ended, it's accepted (KAFKA-19999).

What was written to __transaction_state

The last part of the same run.sh opens the coordinator's log with kafka-dump-log.sh --transaction-log-decoder. It finds and prints whichever of the 50 partitions is non-empty.

=== __transaction_state partition that holds u12-txn
__transaction_state-23
{"producerId":0,"previousProducerId":0,"producerEpoch":0,"transactionTimeoutMs":60000,"transactionStatus":"Empty","transactionPartitions":null}
{"producerId":0,"previousProducerId":0,"producerEpoch":0,"transactionTimeoutMs":60000,"transactionStatus":"Ongoing","transactionPartitions":[{"topic":"u12-txn","partitionIds":[0]}],"clientTransactionVersion":2}
{"producerId":0,"previousProducerId":0,"lastProducerEpoch":0,"producerEpoch":1,"transactionTimeoutMs":60000,"transactionStatus":"PrepareCommit","transactionPartitions":[{"topic":"u12-txn","partitionIds":[0]}],"clientTransactionVersion":2}
{"producerId":0,"previousProducerId":0,"lastProducerEpoch":0,"producerEpoch":1,"transactionTimeoutMs":60000,"transactionStatus":"CompleteCommit","transactionPartitions":[],"clientTransactionVersion":2}
{"producerId":0,"previousProducerId":0,"lastProducerEpoch":0,"producerEpoch":1,"transactionTimeoutMs":60000,"transactionStatus":"Ongoing","transactionPartitions":[{"topic":"u12-txn","partitionIds":[0]}],"clientTransactionVersion":2}
{"producerId":0,"previousProducerId":0,"lastProducerEpoch":1,"producerEpoch":2,"transactionTimeoutMs":60000,"transactionStatus":"PrepareAbort","transactionPartitions":[{"topic":"u12-txn","partitionIds":[0]}],"clientTransactionVersion":2}
{"producerId":0,"previousProducerId":0,"lastProducerEpoch":1,"producerEpoch":2,"transactionTimeoutMs":60000,"transactionStatus":"CompleteAbort","transactionPartitions":[],"clientTransactionVersion":2}
{"producerId":0,"previousProducerId":0,"lastProducerEpoch":1,"producerEpoch":2,"transactionTimeoutMs":60000,"transactionStatus":"Ongoing","transactionPartitions":[{"topic":"u12-txn","partitionIds":[0]}],"clientTransactionVersion":2}
{"producerId":0,"previousProducerId":0,"lastProducerEpoch":2,"producerEpoch":3,"transactionTimeoutMs":60000,"transactionStatus":"PrepareCommit","transactionPartitions":[{"topic":"u12-txn","partitionIds":[0]}],"clientTransactionVersion":2}
{"producerId":0,"previousProducerId":0,"lastProducerEpoch":2,"producerEpoch":3,"transactionTimeoutMs":60000,"transactionStatus":"CompleteCommit","transactionPartitions":[],"clientTransactionVersion":2}

(For each record only the JSON after payload: is kept; the two timestamp fields that change every run and the outer {"version":"1","data":…} wrapper were removed with sed.)

It follows the state diagram in section 2 exactly: one Empty line, then Ongoing → Prepare* → Complete* per transaction. The epoch goes up on the Prepare* line (lastProducerEpoch keeps the previous value), and the partition list is cleared at Complete*. Arrows 2, 6, 10 and 14 in the section 3 diagram are these records. u12-txn went to __transaction_state-23 because the absolute value of "u12-txn".hashCode() modulo 50 is 23, which is the same every run. kafka-transactions.sh shows the same state.

=== kafka-transactions.sh list / describe
TransactionalId	Coordinator	ProducerId	TransactionState	
u12-txn        	1          	0         	CompleteCommit  	
CoordinatorId	TransactionalId	ProducerId	ProducerEpoch	TransactionState	TransactionTimeoutMs	CurrentTransactionStartTimeMs	TransactionDurationMs	TopicPartitions	
1            	u12-txn        	0         	3            	CompleteCommit  	60000               	1794026921817                	31955                	               	

(The start time and duration differ from run to run.) describe reads the coordinator's in-memory state (TransactionMetadata) as-is with a DescribeTransactions request (TransactionCoordinator.handleDescribeTransactions(), from line 312).

5. The reading side — the LSO and the list of aborted transactions

The broker serves only up to the LSO

Look again at this line from example 1's output.

  while B is open: latest offset read_uncommitted=7 (high watermark), read_committed=4 (LSO)

Transaction B's records a1–a3 (offsets 4–6) are already in the log and replicated (HW 7). But no decision has been made, so the LSO is stuck at B's first offset, 4. That is exactly UnifiedLog's definition.

    public long lastStableOffset() {
        // cache the first unstable offset metadata to avoid a concurrent update breaking the isPresent check
        Optional<LogOffsetMetadata> firstUnstableOffsetMetadataCopy = firstUnstableOffsetMetadata;
        if (firstUnstableOffsetMetadataCopy.isPresent() && firstUnstableOffsetMetadataCopy.get().messageOffset < highWatermark()) {
            return firstUnstableOffsetMetadataCopy.get().messageOffset;
        } else {
            return highWatermark();
        }
    }

storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java lines 678–686

The first unstable offset is "the first offset of the earliest transaction still open", computed by ProducerStateManager. If there is no open transaction, or that offset is at or above the HW, the LSO equals the HW. UnifiedLog.read() from part 8 picks the upper limit of a read according to the isolation level.

        LogOffsetMetadata maxOffsetMetadata = switch (isolation) {
            case LOG_END -> localLog.logEndOffsetMetadata();
            case HIGH_WATERMARK -> fetchHighWatermarkMetadata();
            case TXN_COMMITTED -> fetchLastStableOffsetMetadata();
        };
        return localLog.read(startOffset, maxLength, minOneMessage, maxOffsetMetadata, isolation == FetchIsolation.TXN_COMMITTED);

storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java lines 1668–1673

LOG_END is for follower fetches (part 9), HIGH_WATERMARK for read_uncommitted consumers, TXN_COMMITTED for read_committed consumers. When the last argument, includeAbortedTxns, is true, LocalLog.read() collects from .txnindex the aborted transactions that overlap the range read and attaches them to the response (addAbortedTransactions(), LocalLog.java lines 582–603).

The consumer does the filtering

This was the most surprising place in this post. The broker doesn't leave aborted records out. Just like the zero-copy in part 8, it sends the segment's bytes as they are and only attaches a list (abortedTransactions) saying "producer id 0's transaction was aborted from offset 4". The one that actually skips them is the consumer's CompletedFetch.

                if (fetchConfig.isolationLevel == IsolationLevel.READ_COMMITTED && currentBatch.hasProducerId()) {
                    // remove from the aborted transaction queue all aborted transactions which have begun
                    // before the current batch's last offset and add the associated producerIds to the
                    // aborted producer set
                    consumeAbortedTransactionsUpTo(currentBatch.lastOffset());

                    long producerId = currentBatch.producerId();
                    if (containsAbortMarker(currentBatch)) {
                        abortedProducerIds.remove(producerId);
                    } else if (isBatchAborted(currentBatch)) {
                        log.debug("Skipping aborted record batch from partition {} with producerId {} and " +
                                        "offsets {} to {}",
                                partition, producerId, currentBatch.baseOffset(), currentBatch.lastOffset());
                        nextFetchOffset = currentBatch.nextOffset();
                        continue;
                    }
                }

clients/src/main/java/org/apache/kafka/clients/consumer/internals/CompletedFetch.java lines 208–224

Going batch by batch, it adds to an "aborted" set the producer ids of aborted transactions that began at or before the batch's last offset (consumeAbortedTransactionsUpTo()). It skips whole transactional batches from those producer ids (isBatchAborted() = a transactional batch whose producer id is in the set), and when it meets that producer id's ABORT marker, it removes the id from the set. The control batch itself is handled on lines 234–240: its records are not handed to the application, only the offset moves on. It can decide per batch instead of per record because a batch always holds records of a single producer id and epoch (the batch header from part 6).

Up to the yellow boxes is the broker; from the blue box on is the consumer. The two red boxes are "what the application never sees".

The last two lines of example 1 are the result.

read_uncommitted consumer got 8 records: c1@0 c2@1 c3@2 a1@4 a2@5 a3@6 c4@8 c5@9 (position 11)
read_committed   consumer got 5 records: c1@0 c2@1 c3@2 c4@8 c5@9 (position 11)

read_uncommitted receives 8, including the aborted a1–a3; read_committed receives only 5. Neither receives the markers (3, 7, 10) — even under read_uncommitted, control batches are filtered on lines 234–240. Both have a next position of 11. It also means seeing gaps in the offsets (3, 4–7, 10) is normal.

Easy to get wrong: the default is read_uncommitted

If you write with transactions and leave the consumer settings alone, aborted records are visible. The default for isolation.level is read_uncommitted.

                                .define(ISOLATION_LEVEL_CONFIG,
                                        Type.STRING,
                                        DEFAULT_ISOLATION_LEVEL,
                                        in(IsolationLevel.READ_COMMITTED.toString(), IsolationLevel.READ_UNCOMMITTED.toString()),
                                        Importance.MEDIUM,
                                        ISOLATION_LEVEL_DOC)

clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java lines 673–678

DEFAULT_ISOLATION_LEVEL is IsolationLevel.READ_UNCOMMITTED.toString() on line 382 of the same file. One more thing: a read_committed consumer can't receive committed records that come after an open transaction holding back the LSO. Just as the LSO was 4 while B was open in the example, a long-open transaction stalls every read_committed consumer of that partition. That's why, once the producer's transaction.timeout.ms (default 60000, ProducerConfig.java lines 579–583) has passed, the coordinator aborts the transaction on its own (TransactionStateManager.timedOutTransactions(), from line 128; the check interval is transaction.abort.timed.out.transaction.cleanup.interval.ms, default 10 seconds).

6. Stopping zombies — epoch fencing

What is a zombie?

Suppose a process running a transactional producer stalls for a long time in GC. Operations tooling assumes it's dead and starts a new process with the same transactional.id. If the old process then wakes up and commits the transaction it was working on, the same work is done twice. Such an old process is commonly called a zombie, and stopping it is called fencing. Kafka's fence is the epoch.

When the new process calls initTransactions(), the coordinator's prepareInitProducerIdTransit() looks at the current state. If a transaction is open, it takes this branch:

        case TransactionState.ONGOING =>
          // indicate to abort the current ongoing txn first. Note that this epoch is never returned to the
          // user. We will abort the ongoing transaction and return CONCURRENT_TRANSACTIONS to the client.
          // This forces the client to retry, which will ensure that the epoch is bumped a second time. In
          // particular, if fencing the current producer exhausts the available epochs for the current producerId,
          // then when the client retries, we will generate a new producerId.
          Right(coordinatorEpoch, txnMetadata.prepareFenceProducerEpoch())

core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala lines 282–288

Reading the Scala — x match { case A => … case B => … } is pattern matching, similar to Java's switch. Here it branches on the value of txnMetadata.state.

After preparing PrepareEpochFence, handleInitProducerId() has the coordinator itself abort the open transaction (endTransaction(..., TransactionResult.ABORT, isFromClient = false, ...), lines 201–208) and returns CONCURRENT_TRANSACTIONS to the new producer. The new producer retries on this error, and the next InitProducerId gets an Empty with the epoch bumped once more from the CompleteAbort state. From then on, anything the old producer sends with the old epoch is rejected.

Example 3: two producers with the same transactional.id

// Example written for this post (not Kafka source code)
// ...
        KafkaProducer<String, String> p1 = producer("u12-p1");
        p1.initTransactions();
        System.out.println("producer 1 after initTransactions: " + pidEpoch(p1));
        p1.beginTransaction();
        p1.send(new ProducerRecord<>(TOPIC, "z1"));
        p1.flush();
        System.out.println("producer 1 sent z1, transaction still open");

        KafkaProducer<String, String> p2 = producer("u12-p2");
        long t0 = System.currentTimeMillis();
        p2.initTransactions();   // same transactional.id — the coordinator fences p1
        System.out.printf("producer 2 after initTransactions: %s (took %d ms)%n", pidEpoch(p2), System.currentTimeMillis() - t0);

        try {
            p1.send(new ProducerRecord<>(TOPIC, "z2"));
            p1.commitTransaction();
            System.out.println("producer 1 commit: succeeded?!");
        } catch (Exception e) {
            System.out.println("producer 1 commit failed: " + e.getClass().getSimpleName());
        }
// ...

Shortened from ex/u12-fencing/Fence.java. Both producers use transactional.id=u12-fence. pidEpoch() is a helper that peeks, via reflection, at the TransactionManager inside KafkaProducer for checking purposes (not code for an application).

=== two producers, one transactional.id
producer 1 after initTransactions: (producerId=0, epoch=0)
producer 1 sent z1, transaction still open
producer 2 after initTransactions: (producerId=0, epoch=2) (took 552 ms)
producer 1 commit failed: InvalidProducerEpochException
producer 2 committed y1: (producerId=0, epoch=3)
read_uncommitted consumer got: [z1@0, y1@2]
read_committed   consumer got: [y1@2]
=== kafka-dump-log: u12-fence-0 batches
baseOffset: 0 lastOffset: 0 baseSequence: 0 producerId: 0 producerEpoch: 0 isTransactional: true isControl: false
baseOffset: 1 lastOffset: 1 baseSequence: -1 producerId: 0 producerEpoch: 1 isTransactional: true isControl: true
 offset: 1 keySize: 4 valueSize: 6 sequence: -1 headerKeys: [] endTxnMarker: ABORT coordinatorEpoch: 0
baseOffset: 2 lastOffset: 2 baseSequence: 0 producerId: 0 producerEpoch: 2 isTransactional: true isControl: false
baseOffset: 3 lastOffset: 3 baseSequence: -1 producerId: 0 producerEpoch: 3 isTransactional: true isControl: true
 offset: 3 keySize: 4 valueSize: 6 sequence: -1 headerKeys: [] endTxnMarker: COMMIT coordinatorEpoch: 0

(How long initTransactions() took differs from run to run. The same run.sh also prints the __transaction_state decoding, but it has the same shape as in example 1, so here are just the states: Empty(0) → Ongoing(0) → PrepareAbort(1) → CompleteAbort(1) → Empty(2) → Ongoing(2) → PrepareCommit(3) → CompleteCommit(3). The number in brackets is the epoch.)

• Producer 2 gets the same producer id 0. What changes is the epoch (0 → 2). Epoch 1 belongs to the ABORT marker (offset 1) the coordinator wrote when it aborted producer 1's transaction.
• Producer 1's z1 is aborted and invisible to the read_committed consumer. Producer 1, unaware that its transaction has already been aborted, sends z2 and fails.
• The failure is named InvalidProducerEpochException, not ProducerFencedException. z2 is not on disk.

The old producer is stopped by raising only the epoch on the same producer id. Offsets, epochs and results all come from Example 3's output above.

Easy to get wrong: why InvalidProducerEpochException?

From the name alone you'd expect fencing to raise ProducerFencedException. Following where z2 was rejected leads to the first lines where the partition leader starts transaction verification.

    public VerificationGuard maybeStartTransactionVerification(long producerId, int sequence, short epoch, boolean supportsEpochBump) {
        synchronized (lock) {
            // Check if the producer epoch is lower than the stored one, and reject early if it is
            ProducerStateEntry entry = producerStateManager.activeProducers().get(producerId);
            if (entry != null && epoch < entry.producerEpoch()) {
                String message = "Epoch of producer " + producerId + " is " + epoch + ", " +
                        "which is smaller than the last seen epoch " + entry.producerEpoch();
                throw new InvalidProducerEpochException(message);
            }

            if (hasOngoingTransaction(producerId, epoch)) {
                return VerificationGuard.SENTINEL;
            } else {
                return maybeCreateVerificationGuard(producerId, sequence, epoch, supportsEpochBump);
            }
        }
    }

storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java lines 887–903

When the ABORT marker (epoch 1) was written, this partition's ProducerStateEntry started remembering epoch 1. z2, arriving with epoch 0, is rejected right here. Even if the leader had taken the path of asking the coordinator, the result would have been the same: when the coordinator answers PRODUCER_FENCED, AddPartitionsToTxnManager is written to translate it into INVALID_PRODUCER_EPOCH (lines 172–175, "Producers expect to handle INVALID_PRODUCER_EPOCH in this scenario"). The comment in ProducerAppendInfo (lines 153–155) explains that since 2.7, Produce responses use InvalidProducerEpoch instead of ProducerFenced so the client can abort the transaction and retry. To catch fencing in an application, you need to handle both exceptions.

What this post leaves out

• The whole TV1 (versions 0 and 1) path — TransactionCoordinator.endTransactionWithTV1() (from line 544), the client-sent AddPartitionsToTxn and AddOffsetsToTxn (TransactionManager's AddPartitionsToTxnHandler line 1599, AddOffsetsToTxnHandler line 1854), and the epoch bump when moving from TV1 to TV2 (clientSideEpochBumpRequired)
• Two-phase commit (2PC) support — transaction.two.phase.commit.enable, prepareTransaction(), keepPreparedTxn: TransactionManager.java lines 370–379, TransactionCoordinator.handleInitProducerId() lines 138–146
• Coordinator movement and state recovery — TransactionCoordinator.onElection()/onResignation() (lines 478 and 497), TransactionStateManager.loadTransactionMetadata() (from line 453), and what happens when the coordinator changes while sending markers (TransactionMarkerRequestCompletionHandler.scala)
• transactional.id expiration and producer id expiration — TransactionStateManager.removeExpiredTransactionalIds() (line 151), transactional.id.expiration.ms (default 7 days), producer.id.expiration.ms (default 1 day, TransactionLogConfig.java line 57)
• Rotating the producer id when the epoch reaches Short.MAX_VALUE — TransactionMetadata.prepareProducerIdRotation() (line 196), isProducerEpochExhausted() (line 300)
• Error recovery in the idempotent producer (renumbering sequences, unresolved sequences) — TransactionManager's partitionsWithUnresolvedSequences and bumpIdempotentProducerEpoch() (lines 673–689)
• The snapshot (.snapshot files) that restores producer state when a broker restarts — ProducerStateManager.takeSnapshot()/readSnapshot(), dumpProducerIdSnapshot() in kafka-dump-log.sh
• The tool to find and forcibly end hanging transactions — kafka-transactions.sh find-hanging/abort (tools/src/main/java/org/apache/kafka/tools/TransactionsCommand.java)
• Kafka Streams' processing.guarantee=exactly_once_v2 (streams/) — built on top of the transactions in this post

Summary

Wrapping up the series

Looking back over twelve posts

The road of one record, once more

The big road we drew in part 1 can now be walked again using the names in the source.

Solid lines are the order one record passes through; dotted lines are what sits on top of that road. Part 12 adds rules at both ends of the road — the sending side and the reading side.

When you call send(), the record goes into a batch in the RecordAccumulator (part 5), receives this post's producer id, epoch and sequence when it is drained, and the Sender thread sends it as the bytes of a Produce request (part 3). The broker's Processor reads those bytes and puts them on the request queue (part 4); a request handler passes it to ReplicaManager, where a transactional batch is first confirmed with the coordinator (part 12), then gets an offset in UnifiedLog and is written to the end of a segment (parts 6 and 7). Who leads that partition was decided beforehand by the controller through the metadata log (part 10). When followers catch up with Fetch and the HW rises, the acks=all response goes out (part 9), and a consumer's Fetch takes the bytes up to the HW — or, for read_committed, up to the LSO — without copying (parts 8 and 12). Which partitions that consumer reads and how far it has read are recorded by the group coordinator in __consumer_offsets (part 11).

Looking back, the design we met most often in this series was a single one: "write state to a log as records, replicate that log, and read it back to build state." Not just user data. Metadata (__cluster_metadata), groups and offsets (__consumer_offsets), and transactions (__transaction_state) all sit on the same log and the same replication, and even the outcome of a transaction is written to partitions as a record called a marker. The source showed us again and again why Kafka is called "a distributed log".

Where it resembles the earlier series

A few lines on where this resembles the other open-source series on this blog — only where it really does.

• Thread model: the Acceptor, Processor and request handlers of part 4 shared the big picture with the Acceptor, Poller and worker threads in the Tomcat series; the difference was "who reads the socket". Handling connections with one event loop per process is covered in the nginx series, and MPMs that mix processes and threads to share connections in the httpd series.
• Logs and transactions: writing to a log first and applying later, and replicating that log, come from the same idea as a database's redo log and replication. InnoDB's transactions and logs are covered in the MariaDB series. Despite the same name, @Transactional in the Spring series is an abstraction that wraps the commit and rollback of a single database connection, while a Kafka transaction is a separate protocol that ties writes to several partitions (and consumer offsets) together with a coordinator and markers.
• Build: Kafka is built with Gradle, and, like the processMessages step we saw in part 1, it has steps that generate code during the build. The inside of Gradle is covered in the Gradle series. From the Spring series where it all started to this one, the way of reading stayed the same — pin a commit, write notes, check them against the source, and run examples.

What this series left out

The Kafka repository still has several big rooms this series didn't enter. Here are just their names and places.

• Kafka Streams (streams/) — a stream processing library on top of topics. exactly_once_v2 uses the transactions in this post
• Kafka Connect (connect/) and MirrorMaker (connect/mirror/, mirror-coordinator/) — frameworks for moving data to and from external systems and other clusters
• Share groups (share-coordinator/, KafkaShareConsumer, core/src/main/java/kafka/server/share/) — queue-style consumption where records, rather than partitions, are shared out
• Tiered storage (the Remote* classes in storage/) — moving old segments to remote storage
• Security (SASL, TLS, ACLs, Authorizer), quotas (server/…/quota/), the Admin client, multiple log directories (JBOD) and ReplicaAlterLogDirsThread

Thank you for reading along for twelve posts. The source keeps changing. The line numbers in these posts are tied to commit ab1e816411, but the way of finding your way — from a setting's name to a constant, and from the constant to the if that uses its value — should still work in the next release.