How does Kafka make a multi-partition transactional write atomic? Describe the coordinator, control records, and Last Stable Offset.

## The actors - **Transaction coordinator**: a broker-side component. Each `transactional.id` is hashed to a partition of the internal `__transaction_state` topic; the leader of that partition is that id's coordinator. It owns the transaction's lifecycle and durable state. - **`__transaction_state`**: a compacted internal topic storing, per transactional.id, the current state (Empty, Ongoing, PrepareCommit, PrepareAbort, CompleteCommit, CompleteAbort), epoch, timeout, and the set of partitions in the transaction. - **Data partitions**: the user topic-partitions actually being written. ## Step by step 1. **initTransactions()** — producer finds its coordinator, gets a producer id + epoch, and the coordinator recovers any prior transaction. 2. **beginTransaction()** — purely client-side. 3. **First write to a partition** — before sending data there, the producer sends `AddPartitionsToTxn` to the coordinator, which records that partition in the transaction's partition set (persisted to `__transaction_state`). This is how the coordinator later knows where to write markers. 4. **send(...)** — data records are appended to the leader of each data partition **immediately**, stamped with producer id + epoch and a sequence number. They are physically in the log but not yet 'committed' from a read_committed viewpoint. 5. **commitTransaction()** — the producer asks the coordinator to commit. The coordinator runs a mini **two-phase commit**: a. Append `PREPARE_COMMIT` to `__transaction_state` (durable intent). b. Send a **`WriteTxnMarkers`** request to each partition's leader, which appends a **commit control record** (a 'transaction marker') into that partition's log. c. Once all markers are acknowledged, append `COMPLETE_COMMIT` to `__transaction_state`. Because the intent is durably logged before markers are written, a coordinator crash mid-commit is recoverable — on recovery it re-drives the markers. 6. **abortTransaction()** — identical but with `PREPARE_ABORT` and **abort** control records. ## Control records (markers) Control records are special messages in the data partition that consumers never deliver to application code; they only signal 'transaction X with this producer id/epoch committed/aborted here'. They let consumers resolve the fate of the preceding transactional data records in that partition. ## Last Stable Offset (LSO) and read_committed - The **LSO** is the highest offset such that all transactions below it are resolved (committed or aborted). It sits just before the earliest **open** transaction on that partition. - A `read_committed` consumer fetches only up to the LSO, so it never sees data from an in-flight transaction. - Brokers also maintain an **aborted transactions index** per segment; on fetch, the broker returns the list of aborted producer-id ranges so the consumer can **skip aborted records** that lie below the LSO. - A `read_uncommitted` consumer ignores all of this and reads to the high watermark, seeing aborted and uncommitted data. ## Why it is atomic across partitions The records on different partitions become *visible* only after their commit markers are written and the LSO advances past them. Since the coordinator writes markers for all involved partitions as one logical operation (driven from durably-logged intent), readers flip from 'see nothing' to 'see everything' — no partial visibility. A failure before PREPARE_COMMIT yields an abort (eventually via timeout); a failure after is re-driven to completion. This is the heart of atomic multi-partition writes. ## Costs / tradeoffs - Extra latency from coordinator round-trips and marker writes (commit is not free). - read_committed consumers can lag at the LSO when long transactions are open. - `__transaction_state` and marker traffic add load; keep transactions short and reasonably sized.