What is the WriteTxnMarkers request, and why must marker delivery be idempotent and epoch-fenced?

## What WriteTxnMarkers is `WriteTxnMarkers` (API key 27) is an **inter-broker** request — clients never send it. The **transaction coordinator** sends it to every broker that leads a partition involved in a transaction during **phase 2** of the commit/abort protocol. Each request entry contains: - `producerId` (PID) and `producerEpoch` — identifying the transactional producer instance. - `transactionResult` — COMMIT or ABORT. - `coordinatorEpoch` — the coordinator's own epoch, so a leader can reject markers from a stale coordinator. - the list of `TopicPartition`s to mark (including `__consumer_offsets` partitions when offsets were part of the transaction). The receiving leader appends a control batch (the marker) to each listed partition and replies. Only after **all** markers across **all** partitions are acknowledged does the coordinator append `CompleteCommit`/`CompleteAbort` to `__transaction_state`. ## Why idempotency is required Phase 2 is a *retryable propagation* of a decision that is already final (the PrepareCommit/PrepareAbort record). Failures force retries: - **Coordinator crash mid-phase-2**: the partition of `__transaction_state` fails over to a new broker. On load, the new coordinator replays the log, finds the transaction in PrepareCommit/PrepareAbort with no Complete record, and **re-sends WriteTxnMarkers** to all participants — including any that already received it. - **Partition-leader change**: the coordinator retries against the new leader. If marker appends were not idempotent, a re-sent marker could append a *second* COMMIT/ABORT control batch, double-advance accounting, or otherwise corrupt the log. So a leader recognizes that the transaction for that (PID, epoch) is already finalized in its log and treats the duplicate as a no-op. This lets the coordinator retry **indefinitely** with at-least-once delivery, achieving effectively-once finalization. ## Why epoch fencing is required The **producer epoch** is a monotonically increasing number assigned per `transactional.id`. Every `initTransactions()` (an `InitProducerId` call) bumps the epoch and aborts any in-flight transaction from the prior epoch. This is how Kafka fences **zombies** — an old producer process that hung (e.g., long GC) and then resumes, unaware a replacement took over its `transactional.id`. - A partition leader tracks the highest producerEpoch it has seen for a PID. - Any write — data or marker — carrying an **older** epoch is rejected with `INVALID_PRODUCER_EPOCH` / fencing errors. - So a zombie cannot append data, and a stale marker (from a delayed/old coordinator path) cannot finalize a transaction it no longer owns. The `coordinatorEpoch` field plays the analogous role for the coordinator: a leader rejects a `WriteTxnMarkers` from a coordinator whose epoch is older than what it has seen, preventing a deposed coordinator from interfering. ## Why both together give atomicity under failure - The **decision** is durable and single (phase-1 append). - **Idempotency** makes phase-2 safe to retry until every partition is marked. - **Epoch fencing** ensures only the legitimate current producer/coordinator can finalize, blocking zombies and split-brain. The net effect: the transaction commits or aborts atomically across all partitions exactly once, no matter how many brokers crash and how many retries occur. This is the foundation of Kafka's exactly-once semantics (EOS) for read-process-write pipelines (KIP-98, KIP-447 for the per-partition producer-id scaling).