How does each engine achieve exactly-once processing in a Kafka pipeline, and what is the scope of that guarantee?

## What 'exactly-once' actually means here Exactly-once-semantics (EOS) does **not** mean a record is physically processed once — failures cause reprocessing. It means the **observable effects** are as if each input record affected the output and state **exactly once**: no duplicates, no lost updates. In Kafka pipelines this is scoped to the **consume → process → produce** loop plus internal state. ## Kafka Streams Set `processing.guarantee=exactly_once_v2` (the `_v2` variant, post Kafka 2.5/3.0, uses a single producer per instance instead of producer-per-task and is far more scalable than the original `exactly_once`). Mechanism: 1. The app reads records, updates **state stores** (RocksDB), and produces output. 2. State-store mutations are also written to **changelog topics**. 3. All output writes, changelog writes, **and the consumer offset commit** are bundled into one **Kafka transaction** using the idempotent/transactional producer (`transactional.id`, `enable.idempotence=true`). 4. On `commit.interval.ms`, the transaction commits atomically; on failure it aborts and the records are reprocessed from the last committed offset. Consumers downstream must set `isolation.level=read_committed` to skip aborted records. Trade-off: lower `commit.interval.ms` (Streams defaults it to 100ms under EOS) trims end-to-end latency but adds transaction overhead. ## Apache Flink Flink decouples **state consistency** from **sink delivery**: - **State**: Flink periodically injects **checkpoint barriers** into the stream (asynchronous barrier snapshotting, a variant of the Chandy-Lamport algorithm). When a barrier flows through every operator, each snapshots its state to the configured **state backend** (e.g., RocksDB) with the snapshot persisted to durable storage (S3/HDFS). This gives exactly-once *state*. - **Sink**: For exactly-once *output* to Kafka, the `KafkaSink` with `DeliveryGuarantee.EXACTLY_ONCE` implements **two-phase commit**: on each checkpoint it *pre-commits* a Kafka transaction; when the checkpoint is globally complete it *commits* it. If the job dies before commit, the transaction is aborted/recovered. You must set a `transactionalIdPrefix` and tune `transaction.timeout.ms` to exceed the checkpoint interval, or transactions expire and you get data loss/duplicates. ## Spark Structured Streaming Spark uses a **checkpoint location** (a directory on durable storage) holding a **write-ahead log of offsets** and a **state store** (versioned, with HDFS/RocksDB backends). On recovery it replays from the last committed batch. Exactly-once *output* requires the sink to be **idempotent** (deterministic batch IDs) or **transactional**; the built-in Kafka sink and file sink support it, but arbitrary `foreach` sinks do not unless you make them idempotent. ## The universal caveat — end-to-end scope EOS is **transitive only across participating systems**. If your processor makes a non-transactional side effect — an HTTP POST, a write to a DB outside the transaction, sending an email — that effect is **at-least-once** and will be duplicated on reprocessing. Designing for end-to-end EOS means making side effects idempotent (dedup keys, upserts) or keeping them inside the transactional boundary. ## Edge cases - **Zombie fencing**: transactional producers fence out stale instances via epoch bumps so a paused-then-resumed old instance can't double-write. - **Transaction timeout vs checkpoint interval** (Flink) is the classic foot-gun: `transaction.timeout.ms` must be larger than the maximum expected time between checkpoints, bounded by the broker's `transaction.max.timeout.ms`. - **read_committed lag**: consumers may see higher end-to-end latency because they only read up to the last stable offset (LSO).