Transactions

Concepts covered: Exactly-once processing

A trip ends and the fare service sends 2 records: the fare to fares and the driver's payout to driver-payouts. Those are 2 topics, so 2 partitions with 2 leaders, usually on 2 brokers. An idempotent producer promises that each record lands once in its own partition and nothing more: a crash after the first ack and before the second send leaves a fare with no payout. A transaction ties the 2 sends together. Every record sent between begin_transaction() and commit_transaction() becomes committed together, or none of them does.

transactional.id=fare-writer-7
enable.idempotence=true
acks=all
max.in.flight.requests.per.connection=5
transaction.timeout.ms=60000

Setting transactional.id is the switch. The client refuses to start unless idempotence is on, acks=all is set and at most 5 requests are in flight, because a transaction is built on the idempotent producer's id and sequence numbers. The id names a role and stays the same across restarts.

producer.init_transactions()
producer.begin_transaction()
producer.produce("fares", key="trip-9001", value="fare=18.40")
producer.produce("driver-payouts", key="driver-512", value="payout=14.72")
producer.commit_transaction()

init_transactions() runs once per producer. It asks any broker where the transaction coordinator for fare-writer-7 is: the coordinator is the leader of the partition of the internal topic __transaction_state that the id hashes to, a normal broker with a normal replicated log. The coordinator assigns producer id 4001 at epoch 0, writes that assignment to its log, and aborts anything an earlier run under the same id left open. begin_transaction() sends nothing: it flips local state. The first produce() to a partition appends the record there and, on a 4.x cluster, that partition's leader tells the coordinator it now belongs to the open transaction. By the time commit_transaction() runs, the coordinator knows the transaction touched fares-2 and driver-payouts-5.

Commit is written twice

commit_transaction() is a 2-phase write. Phase 1: the coordinator appends a PrepareCommit record for producer 4001 to __transaction_state, a topic created with a minimum in-sync count of 2, so the decision is on 2 disks before anything else happens, and the producer's call returns at this point. Phase 2: the coordinator sends a COMMIT marker to the leader of every partition in the transaction. Each leader appends the marker as a control record, a record with no key and no value that takes 1 offset of its own. When every marker is acknowledged, the coordinator appends CompleteCommit and the transaction is over.

The fare was appended at offset 40 the moment the producer sent it, long before the commit. The marker at 41 is what makes offset 40 committed. The 2 markers land on 2 brokers at 2 different moments, so atomic means that both partitions reach the same outcome, never that they reach it at the same instant. A reader that asks for committed records only holds back at a record until its marker arrives, and skips aborted records for good; how that reader works is a later lesson. A reader with the default setting ignores markers and sees the fare the moment offset 40 exists.

Abort leaves the record in the log

abort_transaction(), or a crash before commit, ends in an ABORT marker written the same 2-phase way: PrepareAbort in the coordinator's log, then a marker in each partition. The fare at offset 40 stays where it is. The log is append-only, so an abort cannot pull a record back; it can only add a marker saying that 40 does not count. Retention deletes the aborted record on the usual schedule, and compaction discards it when the cleaner reaches its segment.

• 50
• 15 min
• 3

One transaction can hold any number of records on any number of partitions and topics, at the cost of 1 marker per partition it touched. A transaction per record on a hot topic doubles the writes to that partition; a transaction per batch of 500 records adds 1 marker. transaction.timeout.ms, 60 seconds by default, bounds how long a transaction may stay open: the coordinator aborts it on its own once the clock runs out, so a producer that waits on a slow external call mid-transaction finds its commit rejected.

About this lesson

This section is part of the Producers: Advanced lesson on DataDriven, the home of data engineering. Lessons cover SQL, Python, Spark, data modeling and pipeline architecture, and each section pairs its explanation with worked examples and practice you run as you go.