Proving exactly-once by killing the stream mid-backlog

Most pipelines I have reviewed claim "exactly-once" somewhere: a README line, an architecture slide, a job title. Very few have a test that fails if the claim stops being true.

Exactly-once is a property of a failure, not of a happy path. To test it you have to cause the failure at an awkward moment, restart, and count. That is more work than a unit test, so it rarely gets written.

So I wrote a script that kills the streaming job with SIGKILL mid-backlog, restarts it, and checks every layer. It runs in CI on every push.

The pipeline

The repo is realtime-lakehouse. The flow is:

• Wikimedia recentchange events arrive over server-sent events.
• A Python producer validates each event against a JSON schema and publishes it to Redpanda. Events that break the contract go to a dead-letter topic.
• Spark Structured Streaming lands the raw records in an Iceberg bronze table, with the Kafka partition and offset kept as columns.
• A second stream parses and deduplicates into silver.edits. A third builds editor sessions.
• dbt on DuckDB reads the Iceberg tables and builds four gold marts. Dagster schedules dbt, runs four asset checks and compacts tables.

Each hop has its own guarantee, and they are not all the same.

Where duplicates and gaps come from

Replay after a crash. A micro-batch reads a range of Kafka offsets, writes files, and commits. If the process dies after the write but before the checkpoint is updated, the restart reads the same range again. A sink that simply appends will write those rows twice.

Offsets against commits. The opposite failure is a gap. If the checkpoint says "batch 12 is done" but the table commit never landed, the restart skips rows that were never stored. The checkpoint and the table commit have to agree, or the sink has to be able to tell that it is ahead of the checkpoint.

A non-idempotent sink and an at-least-once source. My producer is at-least-once on purpose. It saves its resume position (Last-Event-ID) only after Kafka acknowledges everything before it. After a producer crash, a few events are sent again. Those carry the same Wikimedia event id, but they are new Kafka messages with new offsets.

So there are two different problems. Replayed batches must not double-write. Resent events must not double-count. I handle them in two different places.

The mechanism the repo uses

For Kafka to bronze, I rely on Spark's checkpoint plus the Iceberg sink. The docstring in streaming_queries.py says:

each query has its own checkpoint, and the Iceberg sink commits a micro-batch atomically and records the batch id, so a batch replayed after a crash is not committed twice.

The bronze query is small:

def start_bronze(spark, bootstrap):
    kafka = (
        spark.readStream.format("kafka")
        .option("kafka.bootstrap.servers", bootstrap)
        .option("subscribe", TOPIC)
        .option("startingOffsets", "earliest")
        .option("maxOffsetsPerTrigger", os.environ.get("LAKEHOUSE_MAX_OFFSETS", "100000"))
        .load()
    )
    return (
        kafka_to_bronze(kafka).writeStream.format("iceberg").outputMode("append")
        .queryName("bronze").trigger(processingTime=TRIGGER)
        .option("checkpointLocation", f"{CHECKPOINTS}/bronze")
        .toTable(f"lake.{BRONZE}")
    )

There is no foreachBatch, no hand-written MERGE and no dedup in bronze. The guarantee comes from how the two halves fit together:

• Before a micro-batch runs, Spark writes its offset range to the checkpoint's offset log. It marks the batch done in the commit log only after the sink's commit returns. A restart re-runs any batch that has offsets but no commit marker, with the same batch id and the same offsets. That closes gaps.
• Iceberg's streaming sink stores the query id and the batch id in the summary of the snapshot it commits. Before committing, it looks up the last batch id that query committed and skips the commit if this one is not newer. So a batch that reached the table before the crash, but not the commit log, is not written twice. That closes duplicates.

I did not write that logic; I use it, and the test is what tells me whether I am using it correctly. It also depends on the checkpoint surviving: the query id lives there, so a lost or reset checkpoint starts a new query that Iceberg cannot match to the old commits.

For silver, the problem is the resent events, and a checkpoint cannot help with those because they are genuinely different Kafka messages. Silver deduplicates on the Wikimedia event id:

def deduplicate_edits(edits_df):
    return edits_df.withWatermark("event_time", WATERMARK).dropDuplicatesWithinWatermark(
        ["event_id"])

WATERMARK is 10 minutes: the newest event time seen, minus 10 minutes. The dedup state for an event id is kept until the watermark has moved 10 minutes past that event's time, which keeps state bounded. A resent copy carries the same event time as the original, so by the time its state is gone the copy is already behind the watermark and is dropped as late rather than written twice. That is my reading of how Spark orders the two steps; the crash test only resends within the window.

The cost is lateness. The watermark drops every event more than 10 minutes behind, duplicate or not, from silver only. Bronze keeps them, and ops.stream_progress counts them in rows_dropped_by_watermark.

So the claim has two parts, and I keep them separate:

• Kafka to bronze: every Kafka offset appears once.
• Bronze to silver: every unique on-time edit appears once, within the watermark.

The crash test

scripts/prove_exactly_once.py runs in its own compose project with no published ports. An overlay file sets small batches (LAKEHOUSE_MAX_OFFSETS 500, a 2 second trigger) so the backlog takes many micro-batches and the kill lands in the middle of it, not after everything is committed.

The scenario, in order:

1. Publish 50 loops of a 400-event recorded sample, each with unique ids.
2. Resend the 400 events of loop 0, as a resume would.
3. Publish 25 contract violations.
4. Wait 12 seconds, then SIGKILL the stream container, and read the committed counts.
5. Restart the same container on the same checkpoint volume.
6. Publish 100 events stamped 30 minutes in the past.
7. Run the verifier.

Trimmed to the part that matters:

publish("base", "--loops", str(LOOPS))
publish("duplicates")
publish("poison", "--count", str(POISON))

time.sleep(12)  # let a few micro-batches commit, then crash mid-backlog
compose("kill", "-s", "SIGKILL", "stream")
at_crash = marker_json(compose("run", "--rm", "--no-deps", "spark", "python", "-m",
    "wiki_lakehouse.reliability.verify_exactly_once", "--counts-only",
    capture=True), "COUNTS ")
total = LOOPS * sample_size + sample_size  # every event reaches bronze, duplicates too
if not 0 < at_crash["bronze_rows"] < total:
    sys.exit("inconclusive: the crash did not land in the middle of the backlog")

compose("start", "stream")  # same container and checkpoint volume
wait_for_log("stream", "started queries", after=1)
publish("late", "--count", str(LATE), "--minutes", "30")
compose("run", "--rm", "--no-deps", "spark", "python", "-m",
        "wiki_lakehouse.reliability.verify_exactly_once",
        "--expect-kafka", str(total + LATE),
        "--expect-silver", str(LOOPS * silver_per_loop),
        "--expect-dlq", str(POISON))

One guard in there is worth the effort. After the kill, the script reads the committed counts and refuses to continue unless bronze has more than zero rows and fewer than the total. If the crash lands before anything is written, or after everything is written, the test reports "inconclusive" instead of passing. A crash test that can pass without crashing anything is worse than none.

The verifier reads the same Iceberg tables with Spark and compares them with Kafka itself:

checks = {
    "kafka_has_expected_messages": observed["kafka_main_messages"] == args.expect_kafka,
    "bronze_has_every_offset_once": observed["bronze_rows"] == observed["kafka_main_messages"]
    == observed["bronze_distinct_offsets"],
    "silver_has_every_event_once": observed["silver_rows"] == args.expect_silver
    == observed["silver_distinct_events"],
    "late_events_dropped": observed["late_events_in_silver"] == 0,
    "contract_violations_in_dlq": observed["kafka_dlq_messages"] == args.expect_dlq,
}

Look at what is compared. Duplicates are bronze_rows against count(DISTINCT kafka_partition, kafka_offset). Gaps are bronze_rows against the number of messages Kafka actually holds, read with an endingOffsets=latest batch read. A gap shows up as fewer rows than offsets; a duplicate shows up as more rows than distinct offsets.

The expectations come from the scenario, not from numbers I typed in. Kafka should hold 50 x 400 + 400 resent + 100 late = 20,500 messages. The script counts 196 edit or new events in the recording, so silver should hold 50 x 196 = 9,800.

The latest CI run on main killed the stream with 4,484 of the 20,400 events published before the crash committed to bronze. After the restart the verifier saw 20,500 Kafka messages, 20,500 bronze rows and 20,500 distinct offsets, 9,800 silver rows and 9,800 distinct event ids, no late events in silver and 25 messages in the dead-letter topic. Bronze had taken 43 commits to get there. Every check passed: 0 duplicates and 0 gaps.

The numbers around it

The same repo has two measured runs.

Freshness, from the benchmark (benchmarks/results.md, 45 seconds per phase, 10 second trigger, local run). Freshness is edit time to Iceberg commit:

Publish rate	In silver	p95
100 ev/s	2,156	20.2 s
1,000 ev/s	21,952	20.2 s
5,000 ev/s	110,152	20.6 s
9,596 ev/s (target was 15,000)	330,652	21.0 s

Every phase had all expected edit and new events in silver. The producer, one Python process, topped out near 10k events/s, so this does not saturate the pipeline.

The AWS run (benchmarks/aws-run.md): one t3.xlarge in ap-southeast-2 with S3 and Glue, live Wikimedia stream, 2 h 10 min. 868,073 recentchange events ingested, 0 duplicate Kafka offsets in bronze and 0 duplicate events in silver, freshness p95 20.0 s measured as processed_at - event_time in Athena, and all 42 Dagster runs green. The estimated cost was about $0.52. This run predates the change that keeps only edit and new in silver, so its counts cover every event type.

The AWS run checked for duplicates over real traffic, but nothing crashed during it. It answers a different question from the crash proof.

Lessons

Make the test refuse to pass vacuously. Any fault-injection test needs a check that the fault happened where you meant it to.

Verify against the source of truth, not against the pipeline. The verifier counts offsets in Kafka and compares them with bronze. If I had compared bronze with my own earlier count, a bug in my counting would have hidden a bug in the pipeline.

Say which hop each guarantee belongs to. "The pipeline is exactly-once" is not a statement I can defend. "Each Kafka offset lands once in bronze" and "each on-time event id lands once in silver" are, and they have different mechanisms.

A reader can lie too. While building the verifier, a polling reader stalled at 5,479 rows. Iceberg's caching catalog expires an entry after it was last accessed, so a verifier that polled every few seconds kept seeing the same snapshot. Readers of tables that other processes write now disable the cache. A wrong verifier would have looked like a pipeline bug.

What the test does not prove

I would rather list these than have a reviewer find them.

• It kills the Spark stream, not the producer. The producer's resume path is imitated by resending loop 0, not exercised by a real producer crash.
• One kill, one crash point. The kill lands somewhere in the backlog, 12 seconds in. I did not kill during the commit itself, repeat the kill several times, or kill Kafka, the catalog or the object store.
• Dedup is bounded by the watermark. The test does not resend an event after its dedup state has expired. Late events are dropped by design and counted, and replay from bronze is manual.
• It assumes the checkpoint survives. The restart reuses the same checkpoint volume. Losing the checkpoint is a different failure, and this test does not cover it.
• Small data, small batches. The test deliberately uses 500-offset batches and a 2 second trigger to force a mid-backlog crash. The default is 100,000 offsets per trigger.
• It proves one scenario. CI shows that this scenario passes on every push. It is not a proof that no scenario fails.
• Gold is not part of this test. A dbt reconciliation test and a Dagster duplicate check cover gold and silver separately.

Closing

The point of the exercise was not the number. It was to turn a sentence in a README into an assertion that can fail, and to write down where it stops being true.

To run it: python scripts/prove_exactly_once.py. It needs Docker and nothing else.

Code, tests, benchmarks and the AWS write-up: https://github.com/tantn89/realtime-lakehouse

Comments

[Launch Gate]

The sentence I'd print is "a test that fails if the claim stops being true". That reframes what exactly-once claims usually are: a README line nothing can falsify. And there's a second payload in running it on every push that you hint at but don't name: the guarantee here is rented, not owned. It lives in Spark's checkpoint protocol and Iceberg's commit dedup, so a version bump that changes either contract quietly revokes it. The kill-test doubles as a freshness check on your dependencies' guarantees, not just your own wiring.

The caveat you left mid-paragraph is the one I'd promote to a heading: the whole mechanism depends on the checkpoint surviving. Losing it starts a new query id Iceberg can't match, and the failure mode is the worst kind - the pipeline keeps running, rows keep landing, every health signal stays green, and the guarantee is silently gone. The SIGKILL test covers crash-with-checkpoint; the correlated real-world failure is node loss taking both at once. Worth either a checkpoint-loss case in the same test or an alert on query-id change, because "exactly-once while the checkpoint lives" is the actual claim the evidence supports.

The two-problems-two-places split is what most exactly-once writeups miss: batch replay is a framework-fit problem, resent events are a semantics problem, and no single mechanism covers both. Naming which layer owns which failure is what makes the claim readable - the common alternative is the word "idempotent" doing work it was never defined to do.

[Tan Tran]

Agreed, and that's the honest scope of the claim: exactly-once while the checkpoint lives. The query id that Iceberg's dedup keys on is stored in the checkpoint, so losing it starts a new query that Iceberg can't match, and nothing turns red.

The SIGKILL test covers crash-with-checkpoint only. A checkpoint-loss case (which should produce duplicates and prove the boundary) or an alert on query-id change would cover the node-loss scenario you describe. I've noted it as a known limit in the README. And yes on "rented, not owned": running the kill test on every push is what would catch a Spark or Iceberg upgrade that quietly changes either contract.

[Launch Gate]

The README note is the right move, and it's worth saying why: the claim and its boundary now ship together, so a reader gets "exactly-once while the checkpoint lives" instead of discovering the qualifier in production. Most guarantees fail in prose before they fail in code.

If you do add the checkpoint-loss case, there's a nice symmetry available: the test would assert that duplicates DO appear. A test that proves the boundary by watching the guarantee fail turns the known-limit from prose into evidence. And it stands guard from the other direction too - if Iceberg's dedup ever keys on something other than query id, the boundary itself has moved, and that test is the one that tells you.