Module 06 · Day 5 · Kafka 4.3

Transactions and exactly-once

A transaction makes several writes succeed or fail together. With it, an application that reads from Kafka and writes to Kafka can give each input exactly one effect, even through crashes. The guarantee is real, and it stops at the edge of Kafka.

Try it: one partition, two kinds of reader

Write records inside a transaction, then commit, abort, or crash. Watch what each reader can see.

partition log

read_uncommitted sees

read_committed sees

in an open transaction committed aborted marker: C commit, A abort

offset 0

What it is and why it exists

Module 02 showed that idempotence stops duplicates from one producer's retries to one partition. Many applications need more. A typical one reads a record from an input topic, computes something, writes results to one or more output topics, and commits its input offset. That is three or more writes. If the application crashes between them, some are done and some are not, and after a restart the input is processed again.

A transaction groups those writes. Either all of them take effect or none does. The group can include records for several partitions and topics, and also the consumer group's offset commit, because an offset commit is itself a record in a Kafka topic.

Readers take part too. A consumer with isolation.level=read_committed does not receive records from a transaction until it has committed, and never receives records from one that was aborted.

Exactly-once in Kafka means this: for a read-process-write application, each input record affects the output topics and the committed offset exactly one time. It does not mean that your code runs once, and it does not cover anything outside Kafka.

The one-sentence version. A transaction writes records and offsets as one all-or-nothing unit, marks the outcome in the log, and readers that ask for committed data skip everything else.

offset 1

Mental model

Records of a transaction are written to the log at once, like any other records. They are not held back somewhere. What comes later is a small extra record, a marker, that says "the transaction of producer X is committed" or "is aborted". Readers decide what to show by looking at the markers.

A read_uncommitted reader, the default, shows every data record as soon as it is replicated. It ignores the outcome.
A read_committed reader stops at the first record of the oldest transaction that is still open. That position is the last stable offset. When the marker arrives, it moves on, and drops the records if the marker says abort.

Use the demo at the top. Write three records and commit: both readers show three. Write two more and press "Producer crashes": the first reader shows five, the second still shows three and waits. That waiting is the price of the guarantee, and it is the first thing that goes wrong in production.

The pattern transactions were built for

Read from topic A, process, write to topic B, and record progress on A. Without a transaction, a crash at the wrong moment gives duplicate output. Step through both versions:

Step through it: read, process, write, with a crash

Output and offset are written

input topic

output, as read_committed sees it

Choose a mode and press the button.

Where the guarantee stops

Covered by a Kafka transaction	Not covered
Records written to any topics in the same cluster.	A row written to a database.
The consumer group's offset commit.	An email, a payment request, any call to another service.
State kept by Kafka Streams in changelog topics (module 07).	Records written to a different Kafka cluster.

If processing has an effect outside Kafka, a retried transaction repeats that effect. For those cases you still need the outbox pattern or an operation that is safe to repeat.

offset 2

How it works internally

The parts

transactional.id: a name you give the producer. It must be stable across restarts of the same logical instance. It is how Kafka recognises "the same producer, restarted".
Transaction coordinator: a broker that tracks the state of each transaction. It is the leader of one partition of the internal topic __transaction_state; the hash of the transactional id picks which of the 50 partitions.
Producer id and epoch: the numbers from module 02. With a transactional id, the same producer id survives restarts and the epoch goes up.
Markers: control records written to each partition that took part. A marker uses one offset.

One transaction, step by step

initTransactions(). The producer finds its coordinator and registers its transactional id. The coordinator raises the epoch. Any older process with the same id is now fenced: its next request is refused. The coordinator also finishes any transaction the old process left open.
beginTransaction(). This only changes the producer's local state.
send(). Records go to the partition leaders as usual, flagged as transactional. The coordinator learns which partitions take part.
sendOffsetsToTransaction(). The consumer's offsets are added to the transaction. They are written to __consumer_offsets but do not count until the commit.
commitTransaction(). The coordinator writes "prepare commit" to its own log. From this moment the outcome is decided, even if the coordinator crashes.
The coordinator sends a commit marker to every partition that took part, including the offsets partition.
The coordinator writes "complete commit". The transaction is finished.

Steps 5 to 7 are a two-phase commit. Step 5 is the decision. Steps 6 and 7 can be repeated safely after a failure, by the same coordinator or by its replacement.

How a read_committed consumer filters

The broker returns records only up to the last stable offset. For aborted transactions, each segment has a small file, the transaction index, that lists the offset ranges of aborted transactions. The broker sends that list with the data, and the consumer drops those records. unverified: the index file was not inspected in the lab

The transaction timeout

If a producer starts a transaction and then disappears, its records would block committed readers for ever. So every transaction has a timeout, transaction.timeout.ms, 1 minute by default. When it passes, the coordinator aborts the transaction and raises the epoch, so the old producer cannot continue. The broker refuses a timeout above transaction.max.timeout.ms, 15 minutes.

The epoch rises with every transaction

In the lab, the producer epoch was 3 after three transactions and 5 after two more. In Kafka 4 the coordinator raises the epoch at the end of each transaction, which closes a gap where a late message from one transaction could be mistaken for part of the next. unverified: the lab showed the numbers; the reason is from memory

What it costs

Each transaction adds a few requests to the coordinator and one marker for each partition. For large transactions the cost is small. For one record in each transaction it is large. The usual practice is to commit a batch every 100 milliseconds to a few seconds. A longer interval gives more throughput and makes committed readers wait longer.

offset 3

Every relevant config

Broker defaults were read from the lab cluster. Client defaults are from the Kafka 4.3 reference pages checked in modules 02 and 03.

Config	Where	Default	What it does	When to change it
`transactional.id`	Producer	null	Turns on transactions and names this producer.	Set a value that is unique for each running instance and the same after a restart, for example the application name plus the instance number.
`transaction.timeout.ms`	Producer	60000 (1 minute)	The coordinator aborts a transaction that stays open longer than this.	Raise it when one batch of processing can take long. Lower it so that a crashed producer blocks readers for less time.
`enable.idempotence`	Producer	`true`	Required for transactions.	Leave it on.
`isolation.level`	Consumer	`read_uncommitted`	Whether the consumer receives records of open and aborted transactions.	Set `read_committed` on every consumer of a topic that is written with transactions. The default gives no protection.
`enable.auto.commit`	Consumer	`true`	Auto-commit of offsets.	Set `false` in a read-process-write application: the offsets must go through the transaction.
`transaction.max.timeout.ms`	Broker	900000 (15 minutes)	The largest timeout a producer may ask for.	Rarely. A higher limit lets one bad client block readers for longer.
`transactional.id.expiration.ms`	Broker	604800000 (7 days)	The coordinator forgets a transactional id that has been idle this long.	Raise it for producers that run less than once a week.
`transaction.abort.timed.out.transaction.cleanup.interval.ms`	Broker	10000 (10 seconds)	How often the coordinator looks for transactions past their timeout.	Rarely.
`transaction.remove.expired.transaction.cleanup.interval.ms`	Broker	3600000 (1 hour)	How often expired transactional ids are removed.	Rarely.
`transaction.state.log.replication.factor`	Broker	3	Copies of `__transaction_state`.	Keep 3. With fewer brokers than this, transactions cannot start.
`transaction.state.log.min.isr`	Broker	2	In-sync minimum for that topic.	Keep 2.
`transaction.state.log.num.partitions`	Broker	50	Partitions of that topic.	Do not change after the cluster has started.
`transaction.state.log.segment.bytes`	Broker	104857600 (100 MiB)	Segment size of that topic.	Rarely.
`transaction.partition.verification.enable`	Broker	`true`	The partition leader checks with the coordinator that a transaction is really open before it accepts transactional records.	Leave it on. It prevents transactions that hang for ever.
`producer.id.expiration.ms`	Broker	86400000 (1 day)	How long a partition remembers an idle producer id for duplicate detection.	Rarely.
`transaction.two.phase.commit.enable`	Broker	`false`	Lets an outside coordinator, such as a database, decide the outcome of a Kafka transaction.	Only for a product that asks for it. unverified
`processing.guarantee`	Kafka Streams	`at_least_once`	`exactly_once_v2` makes Streams use transactions for you.	Module 07. unverified default

offset 4

Failure scenarios

A producer dies with a transaction open

What breaks: Committed readers of every partition in that transaction stop at its first record. Later records from other producers wait behind it.
You observe: Lag grows for read_committed consumers while read_uncommitted consumers keep up. kafka-transactions.sh describe shows the state Ongoing. In the lab: 68 records visible to one reader, 30 to the other.
What happens next: After transaction.timeout.ms the coordinator aborts it. In the lab, with a 40-second timeout, the state changed to CompleteAbort about 45 seconds after the transaction began: the timeout plus the coordinator's 10-second check. If the producer restarts sooner with the same transactional id, initTransactions() aborts it at once.
Fix: Keep the timeout as low as your processing allows. Restart crashed instances quickly with the same id.

The producer is fenced

You observe: A "producer fenced" or "invalid producer epoch" error on send or commit.
Cause: Another process started with the same transactional id, or the coordinator aborted this producer's transaction after a timeout.
Fix: This error is final for that producer object. Close it and exit or create a new one. If two live instances share an id by mistake, they will fence each other in turn: give each instance its own id.

Processing takes longer than the timeout

What breaks: The coordinator aborts the transaction while your code is still working. The commit fails, the batch is processed again, and it times out again.
Fix: Make batches smaller with max.poll.records, or raise transaction.timeout.ms. Both limits apply: the batch must also finish within max.poll.interval.ms.

A consumer shows aborted data

What breaks: A downstream system receives records that were rolled back.
Cause: The consumer uses the default read_uncommitted. Transactions on the producer side protect nobody unless readers ask for committed data.
Fix: Set isolation.level=read_committed. Check every consumer of the topic, including connectors and tools.

Duplicates in an external system

What breaks: The application sends an email or writes a database row during processing, the transaction aborts, and the retry does it again.
Cause: The transaction covers only Kafka.
Fix: Move the outside effect behind a topic: write a "send email" record in the transaction, and let a separate consumer send it with an id that the receiver can use to ignore repeats.

The transaction coordinator is not available

What breaks: Producers cannot start or commit transactions. Plain producers are not affected.
Cause: The partition of __transaction_state for that id has no leader or is below its in-sync minimum.
Fix: Treat it like any partition in that state (module 04). This is why the internal topic needs 3 copies.

Record counts and offsets do not match

You observe: A topic's end offset is higher than the number of records consumers received. In the lab: 30 records, end offset 33.
Cause: Each marker uses an offset. Aborted records also use offsets and are not delivered.
Fix: Nothing is wrong. Do not use offset arithmetic to count records on transactional topics.

offset 5

Hands-on lab

Every command below was run on 4 October 2026 against the lab cluster. The performance tool can write in transactions, so no code is needed.

k() { docker exec -i kafka-1 /opt/kafka/bin/"$@"; }
B=kafka-1:19092
k kafka-topics.sh --bootstrap-server $B --create --topic payments --partitions 1 --replication-factor 3

Step 1. Write 30 records in three transactions

Ten records each second, and a commit every second.

k kafka-producer-perf-test.sh --topic payments --num-records 30 --record-size 20 --throughput 10 \
  --transactional-id pay-1 --transaction-duration-ms 1000 --command-property bootstrap.servers=$B
k kafka-get-offsets.sh --bootstrap-server $B --topic payments

30 records sent, 9.9 records/sec …
payments:0:33

Thirty records, but the end offset is 33. The three extra offsets are the commit markers.

Step 2. See the transaction in the log

k kafka-dump-log.sh --files /tmp/kraft-combined-logs/payments-0/00000000000000000000.log | grep baseOffset | head -3
k kafka-dump-log.sh --print-data-log --files /tmp/kraft-combined-logs/payments-0/00000000000000000000.log | grep endTxnMarker

baseOffset: 0 lastOffset: 1 count: 2 baseSequence: 0 … producerId: 1000 producerEpoch: 0 isTransactional: true isControl: false
baseOffset: 2 lastOffset: 2 count: 1 baseSequence: 2 … producerId: 1000 producerEpoch: 0 isTransactional: true isControl: false
…
| offset: 12 … endTxnMarker: COMMIT coordinatorEpoch: 0
| offset: 24 … endTxnMarker: COMMIT coordinatorEpoch: 0

Data batches carry isTransactional: true. The markers sit at offsets 12, 24 and 32.

Step 3. Ask the coordinator

k kafka-transactions.sh --bootstrap-server $B list
k kafka-transactions.sh --bootstrap-server $B describe --transactional-id pay-1

TransactionalId  Coordinator  ProducerId  TransactionState
pay-1            1            1000        CompleteCommit

CoordinatorId  TransactionalId  ProducerId  ProducerEpoch  TransactionState  TransactionTimeoutMs
1              pay-1            1000        3              CompleteCommit    60000

Broker 1 is the coordinator for this id. The epoch is 3 after three transactions.

Step 4. Abandon a transaction

Start a producer whose transaction would last 5 minutes, with a timeout of 40 seconds, and kill it after 8 seconds without any clean-up.

docker exec -d kafka-2 /opt/kafka/bin/kafka-producer-perf-test.sh --topic payments \
  --num-records 100000 --record-size 20 --throughput 5 --transactional-id pay-2 \
  --transaction-duration-ms 300000 --command-property bootstrap.servers=$B transaction.timeout.ms=40000
# wait 8 seconds
docker exec kafka-2 sh -c "pkill -9 -f 'transactional-id pay-2'"

k kafka-transactions.sh --bootstrap-server $B describe --transactional-id pay-2
for iso in read_uncommitted read_committed; do
  k kafka-console-consumer.sh --bootstrap-server $B --topic payments --from-beginning \
    --isolation-level $iso --timeout-ms 5000 2>&1 | tail -1
done

CoordinatorId  TransactionalId  ProducerId  ProducerEpoch  TransactionState  TransactionTimeoutMs  …  TopicPartitions
3              pay-2            2000        0              Ongoing           40000                    payments-0

read_uncommitted:  Processed a total of 68 messages
read_committed:    Processed a total of 30 messages

The dead producer wrote 38 records. The default reader received them. The committed reader stopped before them, and would also not see anything written after them by other producers.

Step 5. Watch the coordinator clean up

# wait until 40 seconds after the transaction began, then:
k kafka-transactions.sh --bootstrap-server $B describe --transactional-id pay-2
k kafka-dump-log.sh --print-data-log --files /tmp/kraft-combined-logs/payments-0/00000000000000000000.log | grep endTxnMarker | tail -1
k kafka-get-offsets.sh --bootstrap-server $B --topic payments

… pay-2  2000  …  CompleteAbort

| offset: 71 … endTxnMarker: ABORT coordinatorEpoch: 0
payments:0:72

read_uncommitted:  68 messages
read_committed:    30 messages

The coordinator wrote an abort marker at offset 71. The committed reader is no longer blocked, and it still receives 30 records: the 38 aborted ones are filtered out for ever. The default reader keeps receiving them. They stay in the log until retention removes the segment.

Step 6. Look at the coordinator's topic

k kafka-topics.sh --bootstrap-server $B --describe --topic __transaction_state | head -1

Topic: __transaction_state  PartitionCount: 50  ReplicationFactor: 3
Configs: compression.type=uncompressed,min.insync.replicas=2,cleanup.policy=compact,segment.bytes=104857600,…

offset 6

Common mistakes and misconceptions

People think	What is true
Exactly-once means my code runs once for each record.	Your code can run several times. What happens once is the committed result in Kafka.
Turning on transactions in the producer is enough.	Consumers must set `read_committed`, or they receive aborted records.
Uncommitted records are kept in a buffer.	They are in the partition log at once. Markers and the reader's isolation level decide what is shown.
A Kafka transaction can include my database.	It covers only one Kafka cluster.
Any unique string works as a transactional id.	It must be the same after a restart. A random id each time gives no fencing, and old transactions wait for the timeout.
Transactions make Kafka slow.	The cost is per transaction, not per record. With batches of hundreds of records it is small.
Idempotence and transactions are the same thing.	Idempotence removes duplicates from retries to one partition. Transactions make writes to many partitions atomic and survive restarts.

offset 7

Staff-level questions

When are transactions the right tool, and when are they not?

Right: a stream processing step that reads from Kafka and writes to Kafka, where a duplicate in the output is a real problem, such as money totals or counts that later steps trust.

Not needed: when consumers can ignore repeats cheaply by an id, or when occasional duplicates do no harm, such as metrics. Not sufficient: when the important effect is outside Kafka. There a transaction gives a false sense of safety.

How do you choose transactional ids for an application with ten instances?

Each instance needs an id that no other live instance uses and that the same instance gets again after a restart. A stable instance number from the deployment system works: application name plus that number.

With a random id for each start, a crashed instance's open transaction is not aborted by its replacement, and blocks committed readers until the timeout. With a shared id, instances fence each other. Kafka Streams manages ids for you.

A team reports that their read_committed consumer is stuck but the topic is growing. What do you check?

An open transaction on that partition. Compare the consumer's position with the partition's last stable offset. List transactions and look for one in state Ongoing for longer than expected; the describe output shows its producer, its start time and its timeout.

If the producer is gone, the coordinator aborts it at the timeout. If the timeout is very long, kafka-transactions.sh has an abort command for an operator to end it. Then fix the cause: a timeout set too high, or instances that restart with new ids.

How does the commit interval trade throughput against latency?

Committed readers see nothing until the commit. So the commit interval is added to the end-to-end delay. A commit every 100 milliseconds keeps delay low and spends more on coordinator requests and markers. A commit every 5 seconds is cheap and makes every reader up to 5 seconds late, and a failure repeats up to 5 seconds of work.

Choose from the delay the readers accept, then check that the transaction rate is acceptable for the coordinator.

Why must offsets be committed through the producer in this pattern?

The output records and the input offset must share one fate. If the consumer commits the offset separately, a crash between the two steps gives either repeated output or skipped input. sendOffsetsToTransaction() puts the offset commit under the same marker as the output.

The call also passes the consumer's group metadata, so that a consumer that was removed from the group in a rebalance cannot commit offsets for partitions it no longer owns.

What does exactly-once cost the cluster?

Extra load on the coordinators and on __transaction_state. One marker write for each partition in each transaction. Memory in each partition leader for producer state. And an operational risk: any open transaction delays every committed reader of that partition, so one misbehaving client can affect others.

offset 8

Self-test

1. What can one Kafka transaction contain?

Records for any topics and partitions in one cluster, and consumer group offset commits.

2. What is a marker?

A control record that the coordinator writes to each partition of a transaction to say commit or abort. It uses one offset.

3. What is the default isolation level of a consumer?

read_uncommitted.

4. What is the last stable offset?

The offset of the first record of the oldest open transaction in the partition. Committed readers read only below it.

5. What does initTransactions() do to an older process with the same transactional id?

It raises the epoch, so the older process is fenced, and it finishes any transaction that process left open.

6. A producer crashes with a transaction open. What ends the transaction?

The coordinator aborts it after transaction.timeout.ms, or a restarted producer with the same id aborts it in initTransactions().

7. Which step of a commit decides the outcome?

The coordinator writing "prepare commit" to its own log. Everything after it can be repeated.

8. A topic has 30 records and its end offset is 33. Why?

Three transactions were committed, and each commit marker uses one offset.

9. Name one effect that a Kafka transaction cannot make exactly-once.

Anything outside Kafka: a database write, an email, a call to another service.

10. Which tool shows open transactions?

kafka-transactions.sh, with list and describe.