streaming-platform: exactly-once clickstream, counted at every layer

Context

streaming-platform is a real-time data platform for the storefront clickstream of a fictional homewares retailer. Shoppers browse, add to cart, check out and pay, and the events arrive the way real ones do: some twice, some out of order, some minutes late, a few not even valid. Contracts sit at the front door, Kafka and Flink process everything exactly once, ClickHouse answers queries a few seconds old, Iceberg keeps the record, and a freshness SLO pages when the pipeline stalls.

It is a from-scratch reference system, not a production deployment. What makes it worth reading is the reconciler. A seeded generator writes a ledger of every event and every fault it injected; after each run, the reconciler counts the run at every layer — Kafka, Flink’s output, ClickHouse, the Iceberg table, the dead-letter topic, the window aggregates — by sequence number. A layer is exact only when nothing is missing, nothing is there twice, and no poison leaked.

Constraints

All nodes are containers on one Docker VM with 8 vCPU and 11.7 GiB, so replication and network latency are local, and a CPU stall on the host slows every node at once. A broker kill here is process death, not a network partition. Each experiment ran three times; load runs lasted 90 seconds, failure runs 90 to 330. Everything is in MEASUREMENTS.md, with the raw reconciliation reports in results/.

Architecture

A generator sends Avro through an idempotent producer into clickstream.raw. A transactional gatekeeper decodes and validates each batch inside one Kafka transaction, writing valid events to events.valid and bad bytes, with a reason, to events.dlq. Flink 1.20 runs two SQL jobs: clean deduplicates on event_id and writes events.clean through a two-phase-commit Kafka sink and an Iceberg table that commits a snapshot per checkpoint; windows computes 10-second tumbling counts and 30-second sessions by event time and routes anything behind the watermark to events.late. ClickHouse reads every topic read_committed into ReplacingMergeTree tables keyed on the event. A probe exporter sends one synthetic event every 5 seconds through the front door and watches for it in ClickHouse; Prometheus turns those probes into multi-window burn-rate alerts.

Kafka is three KRaft nodes, each broker and controller, with every topic replicated three times and min.insync.replicas=2. Checkpoints run every 5 seconds to S3 on SeaweedFS.

In the code

The window counts were the one layer that did not stay exact, and this split is why: the late filter routes on the same condition the window uses to drop, which keeps every event in exactly one place only while disorder stays inside the 5-second watermark. Catching up after a failure reorders partitions beyond that bound, so on-time events land in events.late, and after a restore CURRENT_WATERMARK is NULL until the first new watermark, so a late event read in that gap is dropped by the window and never routed.

CREATE TABLE events_clean (
  -- …
  -- Out-of-order tolerance: an event up to 5 s behind the newest one seen is on time.
  WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
-- …
  INSERT INTO agg_windows
  SELECT window_start, window_end, event_type,
         COUNT(*), SUM(COALESCE(amount_cents, 0)), COUNT(DISTINCT session_id)
  FROM TABLE(TUMBLE(TABLE shopper_events, DESCRIPTOR(event_time), INTERVAL '10' SECOND))
  GROUP BY window_start, window_end, event_type;
  -- …
  -- The window operator drops a record when the watermark has reached the last
  -- millisecond of its window (window_end - 1 ms), not merely its own timestamp.
  -- Route on the same condition so that every event lands in exactly one place.
  INSERT INTO events_late
  SELECT event_id, event_type, event_time, CURRENT_WATERMARK(event_time)
  FROM shopper_events
  WHERE CURRENT_WATERMARK(event_time) IS NOT NULL
    AND TIMESTAMPADD(SECOND, 10 - MOD(CAST(EXTRACT(SECOND FROM event_time) AS INT), 10),
          CAST(FLOOR(event_time TO SECOND) AS TIMESTAMP(3))) - INTERVAL '0.001' SECOND
        <= CURRENT_WATERMARK(event_time);

The whole job, including the parallelism and idle-timeout settings that each fixed an earlier miscount, is in flink/sql/20-windows.sql.

Decisions and trade-offs

Results

In steady state, nine load runs and 2.7 million unique events with every fault injected came out exact at every layer, window counts included. ClickHouse made events queryable at a p99 of 6.1, 6.6 and 9.1 s at 1,000, 3,000 and 6,000 events/s, with the median just above half the checkpoint interval at every rate. Five dashboard queries on 6 million rows answered at a p50 of 8 to 26 ms while ingesting 2,000 events/s.

Through three controller-broker kills, three TaskManager kills and three 150-second stalls, every event layer stayed exact. The idempotent producer saw no failed sends; writes paused for about 10 seconds until the dead broker left the in-sync replicas. Restarts rewrote 658 to 7,522 records that aborted transactions had already written, and no read_committed reader saw them. A frozen pipeline paged in 59 to 62 seconds. The dead-letter topic held exactly the injected poison in every run of every experiment.

The window aggregates did not hold. Every failure moved on-time events into the late side output, up to 40% of one broker-kill run, and two restores lost late events from both the windows and the side output. The analysis takes that apart.

What the measurements refused to show

The aggregates are not exact under failure. Catching up after a stall reads partitions at different speeds, so events.clean carries more disorder than the 5-second watermark bound allows. Restoring the windows job also drops late events that arrive before its first new watermark. The raw table and the lake keep every event; the window tables are fast and approximate during incidents.

It got slower with age. After about four hours of experiments, two 1,000/s runs had a p99 of 23 to 25 seconds instead of 6, with checkpoints stretching to 32 seconds and SeaweedFS at its 512 MB memory cap. They were still exact. It is not root-caused.

The gatekeeper fix is unmeasured. The 70-second stall came from librdkafka’s 30-second connection-setup timeout to the dead coordinator. The timeout is now 5 seconds, and that change has not been re-run.

Building it surfaced bugs outside my code too. A broker healthcheck that started a JVM every 5 seconds held each broker at about 150% CPU, which voided every measurement taken before it was fixed. Kafka 4.1.0’s group coordinator wedged after the VM clock stepped back (KAFKA-19888). And librdkafka 2.16.0 aborted inside a timed-out admin request during broker kills (librdkafka#4605).

What I’d do differently

Run the failures before choosing the window semantics. Append-only windows with a late side output were designed for clients that send late. A failure creates a different kind of lateness — disorder introduced by recovery itself — and that lateness looks identical in the output. Watermark alignment on the deduplicating job’s source, and late events emitted from the window operator itself, are now on the roadmap.

Soak, not sprint. Ninety-second runs on a fresh stack are a best case. The four-hour slowdown only showed up because I repeated the load runs at the end.

Give the broker test a real network. QEMU VMs with tc would add the case containers cannot produce: a partition, where the broker is alive but unreachable.