Exactly-Once Streams Job with an External Side Effect
A Kafka Streams app (processing.guarantee=exactly_once_v2) reads clicks, computes 1-minute tumbling counts per campaign_id with a 10 s grace period, calls peek() to POST each result to an external ranking API, and writes to clicks-per-minute. Cluster is 3 brokers, transaction.state.log.replication.factor=3, min.insync.replicas=2. Two downstream teams consume clicks-per-minute with plain Java consumers using default config; they report counts that sometimes appear and are later replaced by a lower value, plus occasional counts inflated roughly 2x after a deploy. The producer's transaction.timeout.ms is at its default and the ranking API has occasional 30 s timeouts.
1. Why do the downstream teams see records that change or vanish?
Read the worked answer
Their consumers use the default isolation.level=read_uncommitted, which lets them read records from transactions that later abort and records beyond the Last Stable Offset. EOS v2 makes the Kafka writes plus offsets atomic - it does not make them invisible to a consumer that ignores transaction boundaries.
Fix: set isolation.level=read_committed on both downstream consumers. They will then only see committed data up to the LSO, and aborted transactions (task migration, rebalance mid-transaction, timeouts) stop leaking through.
The 'value replaced by a lower one' observation is expected behaviour even then: windowed aggregations emit a result when the window closes and again when late-but-in-grace records arrive, keyed by the same window key. Downstream must treat clicks-per-minute as an upsert stream keyed by (campaign_id, window_start), not as an append-only table. If append-only is required, the topology needs suppress(untilWindowCloses) with a matching grace period.
2. Why does the ranking API see ~2x duplicate posts after a deploy, and how do you fix it without abandoning EOS?
Read the worked answer
The POST happens in peek(), which is inside the topology. On rebalance, task migration, or a transaction abort, Streams reprocesses the records from the last committed offset, so the same window result is posted again. EOS covers Kafka reads, Kafka writes and offset commits; it cannot cover an HTTP call, because the external service cannot be enlisted in the Kafka transaction.
Options, best first:
- Idempotency at the service: send a deterministic key (
campaign_id + window_start + window_end) and have the API upsert on it. Duplicate delivery becomes harmless; no topology change. - Outbox into Kafka: write the result to a Kafka topic, then have a separate consumer (with its own retry/DLQ) call the API. Keeps the topology pure Kafka-to-Kafka, keeps EOS meaningful for the Kafka half, and isolates a flaky API from Streams.
- Only as a last resort, a local dedupe store - it will not survive task migration and should not be trusted alone.
Also consider removing peek() entirely: side effects in a topology also hold the producer transaction open, which matters for the next question.
3. What transactional failure mode does the 30 s API timeout create, and what do you change?
Read the worked answer
The side effect occurs inside the records' transaction. While peek() blocks for 30 s the transaction stays open. transaction.timeout.ms defaults to 60000 ms on the producer, and the broker enforces transaction.max.timeout.ms (default 900000 ms). A couple of slow calls can push a single transaction past its timeout; the coordinator then aborts it and marks the producer's ongoing work as aborted, which is retried and can cascade into repeated aborts during the API's bad periods.
Changes:
- Move the HTTP call out of the hot path (outbox pattern above) so transaction lifetime is bounded by Kafka I/O only.
- If the call must stay, keep the transaction short: lower
commit.interval.ms(default 100 ms, so this is probably already fine) and cap API timeouts well belowtransaction.timeout.ms. - Keep
transaction.timeout.msbelowtransaction.max.timeout.ms; if you raise it, raise the broker-side limit in the same change, on every broker, before the app restarts. min.insync.replicas=2withacks=allis non-negotiable for EOS; a single broker loss must not lose committed transactions.