Event Time vs Processing Time (Watermarks & Late Data).

🐞The symptom

A streaming job counts checkout events into hourly buckets.

• 08:00–09:00 bucket, read at 09:01 → 41,802
• Same bucket, read again at 09:20 → 41,802
• Nightly batch job over the same raw topic → 43,115

~3% of events gone, every hour, only in the stream. No errors. No exceptions. Consumer lag: 0.

🔍Step 1 — look at one missing row

event_id=c81  event_ts=08:59:41  kafka_ts=09:00:12

It arrived 31s after the hour. Mobile client on a train, buffered offline, flushed late.

🔍Step 2 — look at the window

stream
  .window(TumblingWindow.of(1 hour))   // ← which clock?
  .aggregate(sum("amount"))

Default in many setups: processing time. So c81 is filed at 09:00:12 → it lands in the 09:00 bucket. Nothing was lost. It was misfiled. Your 08:00 total is low AND your 09:00 total is contaminated.

💡The concept

Every event carries two timestamps:

• event time — when it happened (08:59:41, stamped by the producer)
• processing time — when your job saw it (09:00:12)

"Revenue for hour 8" is an event-time question. Processing time can only answer "what did my pipeline do."

Switch to event time and buckets become correct — but now: when do you close the 08:00 window? You can't wait forever.

🌊Watermarks

A watermark is the job's claim: "I believe I've seen everything with event_time ≤ T." Usually max_seen_event_time − allowed_lateness.

.assignTimestamps(e -> e.event_ts)
.withWatermark(outOfOrderness = 2 minutes)

At ~09:02 the watermark crosses 09:00 → the 08:00 bucket fires. Anything after that is late.

⚠️Three things practitioners get wrong

1. Thinking event time removes the trade-off. It relocates it. More allowed lateness = more correct, more latency. 2 min lateness means your hourly dashboard is permanently 2 min stale.
2. Dropping late data silently — the default almost everywhere. Emit a counter for late events BEFORE you tune. Pick lateness from the real p99 of (kafka_ts − event_ts), not from a round number that felt nice.
3. Trusting client clocks. Phone clocks skew by minutes. One event stamped 2027 drags the watermark to 2027 and instantly closes every open window — watermark poisoning, and it's unrecoverable in-flight. Clamp on ingest:

event_ts = min(client_ts, ingest_ts)
reject if client_ts > ingest_ts + 1h

🔁The part that bites the dashboard

If late data exists, a "closed" bucket can still change. You get two honest choices:

(a) make the sink correctable — upsert keyed by window_start, or emit retractions

(b) declare the bucket final and accept a known ~0.1% undercount

The original bug was picking neither — late events were silently discarded into a bucket nobody could fix.

🎯Your exercise

Over last week's raw events, run:

SELECT percentile_cont(0.99)
  WITHIN GROUP (ORDER BY kafka_ts - event_ts)
FROM events;

Then answer two things:

1. With allowedLateness = 1 minute, what % of your events would be dropped?
2. Is your sink keyed by window_start — i.e. could a correction even be applied, or would it double-count?