04 - Exactly-Once Semantics in Kafka: Transactions and Coordination

The service that processes an order and fails halfway

You have a service that reads orders from a topic, enriches them and writes the result to another topic. For each batch it does three things in a row: it writes the output, saves the offset it read, moves on to the next batch. If the process dies between the first and the second, on restart it repeats the batch and the output appears twice. If you swapped the steps (offset first, then output) a crash in the middle would lose the batch. With two separate operations there is no right order: you need a way to make them become one.

This is what Kafka transactions offer, and it is the heart of what the documentation calls exactly-once semantics. The complete formula requires four elements working together: an idempotent producer, a transactional.id, a consumer configured with isolation.level=read_committed and enable.auto.commit=false, and the writing of the consumer's offset inside the transaction. It holds only for data that stays inside Kafka: an external database is not part of it, and we will see this at the end. Facts and configuration values come from the Kafka 4.4 documentation (consulted on 2026-11-12) and from the client's Javadoc; the behaviours I reproduced with Java 21 models and with the Mock* classes of the Kafka client 4.4.0, without a broker.

Why two separate operations are not enough: the model

The following model compares two versions of the same service. It processes three input records (offsets 0 to 2) and fails before saving the offset. On restart it reprocesses from the saved offset, which is still 0. The difference is whether the outputs of the first attempt are already visible to a reader or have stayed inside a transaction that was never completed.

import java.util.*;

public class Atomicity {
    // Processes three input records and writes three outputs; the process fails before the confirmation.
    // Then it restarts from the saved offset and reprocesses. What does a reader of the output see?
    static List<String> run(boolean transactional) {
        List<String> output = new ArrayList<>();         // output visible to readers
        List<String> inTransaction = new ArrayList<>();  // written, not yet visible
        final long savedOffset = 0;

        // First attempt: writes the outputs...
        for (long o = savedOffset; o < 3; o++) {
            if (transactional) inTransaction.add("out-" + o); else output.add("out-" + o);
        }
        // ...and fails BEFORE saving the offset (and, in the transactional case, before the commit).
        if (transactional) inTransaction.clear();        // the transaction is aborted: the records stay hidden

        // Restart: starts again from 0 because the offset was not saved.
        for (long o = savedOffset; o < 3; o++) output.add("out-" + o);
        return output;
    }

    public static void main(String[] args) {
        System.out.println("without transaction: " + run(false));
        System.out.println("with transaction:    " + run(true));
    }
}

// Output:
// without transaction: [out-0, out-1, out-2, out-0, out-1, out-2]
// with transaction:    [out-0, out-1, out-2]

Without a transaction the reader sees six records: the three of the first attempt plus the three reprocessed. With the transaction the first attempt is aborted (the line inTransaction.clear() represents the abort) and the reader sees only the three records of the second. The model is deliberately reductive: it does not show how Kafka makes those records invisible. That is done by the markers, and we will see them shortly. The point to fix here is the condition: output and offset must be in the same transaction, otherwise there is no atomicity.

The transactional API, in the order in which it must be called

The KafkaProducer Javadoc fixes the order: with transactional.id set, initTransactions() must be called once before any other operation; then, for each transaction, beginTransaction(), the send() calls, sendOffsetsToTransaction() and finally commitTransaction() or abortTransaction(). A transactional producer has only one open transaction at a time and all its sends must be inside a transaction.

The listing below walks through the sequence with the MockProducer of kafka-clients-4.4.0.jar (Maven Central, SHA-1 verified): it is real library code, but it is a test double, it does not talk to a broker. What it demonstrates is the API contract, not the distributed semantics. In real life the offset comes from the consumer with consumer.groupMetadata(), and here a MockConsumer provides it.

import org.apache.kafka.clients.consumer.MockConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.producer.MockProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Map;

public class TransactionalApi {
    public static void main(String[] args) {
        MockProducer<String, String> producer =
            new MockProducer<>(true, null, new StringSerializer(), new StringSerializer());

        try {
            producer.beginTransaction();                       // without initTransactions()
        } catch (IllegalStateException e) {
            System.out.println("beginTransaction before initTransactions: " + e.getMessage());
        }

        producer.initTransactions();
        Map<TopicPartition, OffsetAndMetadata> offsets =
            Map.of(new TopicPartition("orders-input", 0), new OffsetAndMetadata(3));
        MockConsumer<String, String> consumer = new MockConsumer<>("earliest");   // provides groupMetadata()

        producer.beginTransaction();                           // transaction 1: succeeds
        producer.send(new ProducerRecord<>("orders-processed", "o1", "ok-1"));
        producer.send(new ProducerRecord<>("orders-processed", "o2", "ok-2"));
        producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata());     // the consumer's position, in the same transaction
        producer.commitTransaction();
        System.out.println("transaction 1 committed: " + producer.transactionCommitted()
            + ", visible records: " + producer.history().size()
            + ", committed offsets: " + producer.consumerGroupOffsetsHistory().size());

        producer.beginTransaction();                           // transaction 2: gets aborted
        producer.send(new ProducerRecord<>("orders-processed", "o3", "ok-3"));
        producer.abortTransaction();
        System.out.println("transaction 2 aborted: " + producer.transactionAborted()
            + ", visible records: " + producer.history().size());
    }
}

// Output:
// beginTransaction before initTransactions: MockProducer hasn't been initialized for transactions.
// transaction 1 committed: true, visible records: 2, committed offsets: 1
// transaction 2 aborted: true, visible records: 2

The first line of output shows the error if you skip initTransactions(): beginTransaction() throws IllegalStateException (the message text is the MockProducer's; the real producer's Javadoc documents the same exception for this case). Transaction 1 sends two records and the input offset 3 with sendOffsetsToTransaction: after the commit the history contains two records and one offset. Transaction 2 sends a third record and aborts it: the visible records stay two, because the aborted one does not enter the history. On a real broker the aborted record is written to the log and hidden by the markers.

For error handling the Javadoc distinguishes two families. ProducerFencedException, OutOfOrderSequenceException and AuthorizationException are not recoverable: the only move is to close the producer. For the other KafkaException you call abortTransaction() and try again. The "Design" section adds a detail that is easy to forget: after an abort you must also move the consumer's position back, because the records of that batch have to be read again. For the more recent error categories (retriable, abortable, fatal) the 4.4 Design page applies.

Markers and whoever reads: read_committed and LSO

A transactional producer writes its records normally into the partition's log, but when the transaction ends the broker adds a COMMIT or ABORT marker. A consumer with isolation.level=read_committed (the default is read_uncommitted) discards the records of aborted transactions and does not go beyond the last stable offset (LSO), that is, the first offset of a transaction that is still open. The consumer documentation says so: with read_committed, poll() returns records "up to the last stable offset".

import java.util.*;

public class Isolation {
    enum Kind { DATA, COMMIT, ABORT }
    record Entry(Kind kind, String txn, String value) {}      // txn == null: non-transactional record

    static List<Entry> log = new ArrayList<>();
    static void data(String txn, String v) { log.add(new Entry(Kind.DATA, txn, v)); }
    static void close(String txn, Kind outcome) { log.add(new Entry(outcome, txn, null)); }

    // LSO: first offset of a transaction that is still open (or the end of the log if there are none).
    static int lso() {
        Set<String> closed = new HashSet<>();
        for (Entry v : log) if (v.kind() != Kind.DATA) closed.add(v.txn());
        for (int i = 0; i < log.size(); i++) {
            Entry v = log.get(i);
            if (v.kind() == Kind.DATA && v.txn() != null && !closed.contains(v.txn())) return i;
        }
        return log.size();
    }

    static List<String> read(boolean readCommitted) {
        Map<String, Kind> outcome = new HashMap<>();
        for (Entry v : log) if (v.kind() != Kind.DATA) outcome.put(v.txn(), v.kind());
        List<String> seen = new ArrayList<>();
        int limit = readCommitted ? lso() : log.size();
        for (int i = 0; i < limit; i++) {
            Entry v = log.get(i);
            if (v.kind() != Kind.DATA) continue;                          // markers never reach the application
            if (readCommitted && v.txn() != null && outcome.get(v.txn()) == Kind.ABORT) continue;
            seen.add(v.value());
        }
        return seen;
    }

    public static void main(String[] args) {
        data(null, "x");                 // 0  non-transactional
        data("T1", "a");                 // 1
        data("T1", "b");                 // 2
        data("T2", "c");                 // 3
        close("T1", Kind.COMMIT);       // 4
        data("T2", "d");                 // 5
        close("T2", Kind.ABORT);        // 6
        data("T3", "e");                 // 7  still open
        System.out.println("read_uncommitted: " + read(false));
        System.out.println("read_committed:   " + read(true) + "  (LSO=" + lso() + ")");

        // A slow transaction blocks readers even for the committed data that follows it.
        log.clear();
        data("T1", "a");                 // 0  T1 stays open
        data("T2", "f");                 // 1
        close("T2", Kind.COMMIT);       // 2  T2 is already committed
        System.out.println("T1 open, T2 committed -> read_committed sees " + read(true) + " (LSO=" + lso() + ")");
        close("T1", Kind.COMMIT);
        System.out.println("after T1's commit         -> read_committed sees " + read(true));
    }
}

// Output:
// read_uncommitted: [x, a, b, c, d, e]
// read_committed:   [x, a, b]  (LSO=7)
// T1 open, T2 committed -> read_committed sees [] (LSO=0)
// after T1's commit         -> read_committed sees [a, f]

In the first scene the log has records from three transactions plus a non-transactional one. A read_uncommitted reader sees everything (x a b c d e); a read_committed one sees only x a b: c and d belong to T2, aborted, while e is from T3, still open, and lies beyond the LSO (7). The second scene is the consequence that weighs in production: T1 is open at offset 0 and T2 has already committed, but the LSO is 0, so read_committed sees nothing, not even f. Only after T1 commits do both appear. A slow or forgotten transaction blocks the readers of that partition, and that is why time limits exist: transaction.timeout.ms (60000 ms by default on the producer) and the broker's ceiling transaction.max.timeout.ms (900000 ms), with which the coordinator aborts by itself transactions that run too long.

The coordinator and zombie fencing

The transaction coordinator runs in the brokers and keeps the state of transactions in the internal topic __transaction_state, which has 50 partitions (transaction.state.log.num.partitions) and, by default, replication factor 3 and transaction.state.log.min.isr 2. That is why the documentation warns that transactions, by default, require a cluster of at least three brokers; in development the two parameters are lowered.

The transactional.id identifies a logical instance of the producer across restarts. When a new instance calls initTransactions() with the same identifier, the coordinator increases the epoch, aborts any transaction left open and, from that moment, rejects the commits of whoever still has the old epoch. It is the protection against the "zombie": a producer that was paused (a garbage collector pause, a network partition) and wakes up after its replacement has already started.

import java.util.*;

public class Fencing {
    static final Map<String, Integer> epoch = new HashMap<>();     // state in the transaction coordinator

    static int initTransactions(String transactionalId) {
        return epoch.merge(transactionalId, 1, Integer::sum);      // every init raises the epoch
    }

    static String commit(String transactionalId, int producerEpoch) {
        return producerEpoch < epoch.get(transactionalId) ? "ProducerFencedException" : "commit accepted";
    }

    public static void main(String[] args) {
        int a = initTransactions("orders-0");                      // producer A starts
        System.out.println("A: epoch " + a);
        int b = initTransactions("orders-0");                      // A stalls, B starts with the same id
        System.out.println("B: epoch " + b);
        System.out.println("A tries to commit: " + commit("orders-0", a));
        System.out.println("B tries to commit: " + commit("orders-0", b));
    }
}

// Output:
// A: epoch 1
// B: epoch 2
// A tries to commit: ProducerFencedException
// B tries to commit: commit accepted

A's epoch is 1, B's is 2. When A, awakened, tries to commit, the coordinator (here a Map) sees epoch 1 lower than 2 and replies with ProducerFencedException; B goes through. The model does not cover the fact that since 4.0 the epoch is incremented at every transaction, not only at every initialization: it is the "server-side defense" protocol of KIP-890, active by default on the server since 4.0 and governed by the transaction.version flag. As a rule for choosing the identifier the Javadoc says that it is usually derived from the identity of the shard of a partitioned, stateful application; the Design section adds to use one producer for each consumer.

Kafka Streams: the same thing with one property

If your flow is a Kafka Streams application you write none of this: you set processing.guarantee to exactly_once_v2 and the library manages the transactional producer, the read_committed consumer and the offsets in the transaction. The listing creates the configuration with the 4.4.0 library (kafka-streams-4.4.0.jar) and prints two things worth knowing by heart: the accepted values and the commit interval.

import org.apache.kafka.streams.StreamsConfig;
import java.util.*;

public class StreamsEos {
    static StreamsConfig config(String guarantee) {
        Map<String, Object> p = new HashMap<>();
        p.put(StreamsConfig.APPLICATION_ID_CONFIG, "order-enrichment");
        p.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");   // not contacted
        if (guarantee != null) p.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, guarantee);
        return new StreamsConfig(p);
    }

    public static void main(String[] args) {
        for (String g : new String[] {null, StreamsConfig.EXACTLY_ONCE_V2}) {
            StreamsConfig c = config(g);
            System.out.println("processing.guarantee = " + c.getString(StreamsConfig.PROCESSING_GUARANTEE_CONFIG)
                + ", commit.interval.ms = " + c.getLong(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG));
        }
    }
}

// Output:
// processing.guarantee = at_least_once, commit.interval.ms = 30000
// processing.guarantee = exactly_once_v2, commit.interval.ms = 100

With the default guarantee the commit happens every 30000 ms; with exactly_once_v2 the library brings the interval to 100 ms, because the commit coincides with the transaction's commit and with the moment the output becomes visible to read_committed readers. A higher interval reduces the number of transactions but lengthens the latency with which data becomes visible: that is the trade-off, and you set it. In 4.4 the accepted values are two, at_least_once and exactly_once_v2; version 1 is gone. For the details of Streams, including the minimum brokers, see the relevant page of the series.

What it costs, and when it is not enough

I do not give you a percentage of lost throughput here: no reliable figure exists outside a measurement on your hardware and your load, and those that circulate without a description of the test should be treated with suspicion. The costs are qualitative but concrete. Every transaction adds exchanges with the coordinator and marker writes, so transactions with more records are preferable. read_committed readers see data later, at the commit. Long transactions block readers. And the most important parameter to tune is the duration: large batches per transaction lower the cost per record and increase the visibility delay.

The most important boundary is that the guarantees end where Kafka ends. The documentation says so about delivery to other systems: that system's cooperation is needed. If your service, after reading an order, writes to PostgreSQL, the write to the database is not part of the Kafka transaction. The two sensible roads are to make the consumer idempotent (a unique key that the database refuses the second time, see the article on idempotency in consumers) or to publish events from the database state with the outbox pattern. Often the first costs less than any transaction.

What you can do now

You can explain why writing the output and saving the offset in two steps can be neither safe nor atomic, and how a transaction joins them. You know the order of calls of the transactional API, which exceptions are fatal and which are resolved with an abort. You can read a log with markers: why a read_committed reader skips aborted records and stops at the LSO, and why a transaction left open for a long time blocks everyone. You know what a transactional.id does and why two instances with the same one do not coexist. You know when to rely on exactly_once_v2 in Streams and that the guarantee does not cross the boundary towards an external database. If you want to explain the duplicates on the other side, those generated by the producer's retries, start from the previous article. If you arrive from another point of the series, the links are: the fundamentals of topics and partitions, Flink and Iceberg.