How does Kafka's transactional producer combine the idempotent producer with a transaction coordinator to give exactly-once semantics for a read-process-write pipeline, and what exactly does that atomicity cover?

## The idempotent producer The idempotent producer solves a narrower but foundational problem: what happens when a producer sends a record, the broker writes it and replies, but the acknowledgment is lost on the network, so the producer's client library retries and sends the same record again? Without protection this creates a duplicate in the log even though nothing 'failed' from the application's point of view. Kafka fixes this by: 1. **assigning every producer instance a unique Producer ID (PID)** when it initializes, and 2. **having the producer tag every record it sends to a given partition** with a strictly increasing per-partition sequence number. The broker keeps the last few sequence numbers it has seen per (PID, partition) and, if a retried write arrives with a sequence number it has already committed, it acknowledges success without writing a duplicate. This is enabled by default in modern Kafka via `enable.idempotence=true` and costs essentially nothing extra — it closes the retry-duplicate hole for a single producer writing to a single partition. ## The transactional producer The transactional producer builds on top of idempotence to solve a bigger problem: atomically writing to multiple partitions or topics, and atomically tying that write to 'the input record has been consumed' in the same all-or-nothing unit. 1. A transactional producer is configured with a stable `transactional.id`, which lets it recover its PID and epoch across restarts. 2. When a transaction begins, the producer registers with a **transaction coordinator**, a designated broker for that `transactional.id`, and writes markers describing which partitions are part of the transaction. 3. As the producer writes records, it can also call `sendOffsetsToTransaction()` to fold the consumer group's offset commit into the same transaction — this is the piece that makes a 'read one, produce many, mark the input as done' pipeline atomic. 4. When the app calls `commitTransaction()`, the coordinator writes a commit decision to its own internal log, then writes a COMMIT control marker into every partition touched by the transaction, including the consumer-offsets partition. 5. If the producer crashes before calling commit, or the coordinator times out waiting, an ABORT marker is written instead. ## Why the protocol exists This exists because 'read a message, transform it, write results, and mark the read message as done' spans several partitions and, without a transaction, there's no way to make all of that atomic — a crash after writing the output but before committing the input offset would redeliver the input and reprocess it, duplicating the output; a crash after committing the offset but before finishing the writes would silently drop output. The transactional protocol collapses that multi-step, multi-partition operation into a single commit/abort decision so consumers only ever see a transaction's writes as a whole, never partially. ## What it costs The cost is real on both sides. - **Latency per transaction.** Every transaction requires coordination round trips with the transaction coordinator, which adds latency per transaction, so throughput is best when transactions batch many records rather than committing one record per transaction. - **Buffer on the consumer side.** `read_committed` consumers must also buffer records from in-flight transactions rather than exposing them immediately, which adds end-to-end latency proportional to how long producers hold transactions open. - **The Kafka boundary.** And critically, this mechanism only covers Kafka topics and Kafka consumer-group offsets — it says nothing about a side effect that leaves Kafka, such as an HTTP call, a write to a separate database, or a message published to a different broker; that gap is exactly why 'exactly-once' in Kafka's own documentation is scoped to Kafka-to-Kafka pipelines, and why teams that need it to reach an external system still need idempotency at that boundary. ## Failure modes and the canonical example - **The zombie producer.** A concrete production failure mode is the 'zombie producer': if a producer instance hangs, for example during a long GC pause, while holding an open transaction, and a second instance starts up with the same `transactional.id`, common after a supervisor restarts a stuck process, the coordinator bumps the producer epoch for that `transactional.id` and fences the old instance — any further writes or commit attempts from the zombie fail, preventing it from committing stale data after the new instance has taken over. - **Transactions left open too long.** Another common issue is transactions left open too long past `transaction.timeout.ms`, which the coordinator proactively aborts; if an application doesn't handle the resulting abort correctly it can silently drop work. Kafka Streams is the canonical worked example: with `processing.guarantee=exactly_once_v2`, every task wraps its input-offset commit, state-store changelog writes, and output records into one transaction per commit interval, giving end-to-end exactly-once for pure Kafka-in/Kafka-out topologies — but the moment that topology calls out to an external sink, such as a REST API or a non-transactional database, the guarantee reverts to at-least-once at that boundary and the application must add its own idempotency there.