Handling Late & Out-of-Order Data

late and out-of-order data is the single failure mode that separates a streaming pipeline which is correct from one that merely runs — and it is the topic senior data engineers most often hand-wave through until a downstream dashboard quietly disagrees with the source of truth by three percent every Monday morning. Every real event stream — clickstream beacons buffered on a phone in airplane mode, IoT readings queued behind a flaky cellular modem, payment events fanned across three regional brokers — delivers records whose timestamps do not march monotonically forward. A sale that happened at 10:00:03 can land in your consumer after a sale that happened at 10:00:07, and a straggler stamped 09:58 can show up a full ten minutes after your 10:00 report already shipped. The engineering question is never "will data arrive late and out of order" — in a distributed system it always will — but "how late is late enough to matter, and what does the pipeline do when it happens."

This guide is the walkthrough you wished existed the first time an interviewer asked "explain event time versus processing time," or "how would you set the watermark delay for a source with a slow partition," or "a late event arrives after the window already fired — walk me through exactly what happens." It moves through the five load-bearing concepts — the difference between event time and processing time, the watermark as a heuristic bound on lateness, the three windowing strategies (tumbling, sliding, session), allowed lateness together with side outputs so nothing is ever dropped silently, and the reordering patterns that buy you exactly-once correctness. Each section pairs a teaching block with a Solution-Tail interview answer — code, a step-by-step trace, an output table, then a concept-by-concept breakdown of why it works.

When you want hands-on reps immediately after reading, drill the streaming practice library →, rehearse on the event-processing practice library →, and sharpen the temporal axis with the time-series practice library →.

---

On this page

• Event-time vs processing-time
• Watermarks — bounding lateness
• Windowing — tumbling, sliding, session
• Allowed lateness & side outputs
• Reordering & correctness patterns
• Cheat sheet — late & out-of-order data recipes
• Frequently asked questions
• Practice on PipeCode

---

1. Event-time vs processing-time

The clock you compute against decides whether late data is a correctness bug or a non-event

The one-sentence invariant: every streaming computation is anchored to one of three clocks — the event time stamped when the thing happened, the ingestion time when it entered the pipeline, or the processing time when an operator saw it — and choosing event time is what turns "a late record" from a silent, unrecoverable corruption into a bounded, well-understood, correctable case. The clock you pick in the first design meeting is not a tuning knob you can flip later; it decides the shape of every window, the meaning of every watermark, and whether "we replayed yesterday and got a different number" is a bug report or expected behaviour. Processing-time pipelines are trivial to build and impossible to make reproducible; event-time pipelines cost more up front and are the only ones whose results survive a replay.

The three clocks — and why only two of them are honest.

• Event time. The timestamp embedded in the record itself, set as close as possible to when the real-world event occurred — the moment a user tapped "buy," the instant a sensor sampled a temperature. It is immutable, it travels with the record, and it is the only clock under which a replay produces the identical result. Event time is what "correct" means.
• Ingestion time. The timestamp the pipeline assigns when the record first enters the system (e.g. when Kafka appends it to a partition). It is monotonic per partition and cheap, but it conflates "when it happened" with "when it arrived," so a phone that buffered events offline for an hour gets an ingestion time an hour late — invisible as lateness, wrong as history.
• Processing time. The wall-clock reading of the operator at the moment it processes the record. It requires no timestamps in the data and gives the lowest latency, but it is non-deterministic: the same input replayed on a faster machine, or after a backfill, lands in entirely different windows. Processing time cannot be reproduced and cannot be reasoned about historically.

The 2026 reality — event time is the default, processing time is the exception.

• Event time is the correct default for any pipeline whose output is compared against a system of record: revenue, billing, SLA compliance, funnel analytics, anything an auditor or a finance team will reconcile. If the number has to match a database tomorrow, it has to be computed in event time today.
• Processing time is legitimate only where latency dominates correctness and no historical comparison exists: a live "requests per second right now" gauge, a coarse liveness heartbeat, an alert that only cares about the last few seconds. The moment someone asks "what was it yesterday at 3pm," processing time has already failed.
• Ingestion time is a pragmatic middle ground when the source genuinely has no trustworthy embedded timestamp — but it is a fallback, and the honest move is to note in the schema that ingestion time is standing in for event time and will misattribute buffered data.

The two vocabularies interviewers conflate — and you must not.

• Late is a statement about the watermark: an event is late if it arrives after the pipeline has already declared its event-time slot complete. Lateness is measured in event time relative to the completeness marker.
• Out-of-order is a statement about arrival sequence: an event is out of order if it arrives after an event with a later event time. Out-of-orderness is about the order records show up, independent of any watermark.
• The trap. All late events are, by definition, out of order — but most out-of-order events are not late. A stream can be wildly out of order and still perfectly correct if the watermark is set to tolerate that disorder. Conflating the two is the single most common way candidates reveal they've never run a real event-time job.

What interviewers listen for.

• Do you say "event time" before the interviewer prompts you — and give the replay-reproducibility reason? — senior signal.
• Do you cleanly separate late (watermark-relative) from out-of-order (arrival-relative)? — required answer.
• Do you name processing time's fatal flaw — non-determinism under replay — rather than just "it's less accurate"? — senior signal.
• Do you flag that ingestion time misattributes buffered data instead of treating it as equivalent to event time? — senior signal.
• Do you describe the pipeline's job as "compute per event-time slot, and decide what to do with stragglers" rather than "process the stream"? — required answer.

Worked example — measuring event-time vs processing-time skew

Detailed explanation. The most useful first artifact in any late-data discussion is the skew between when events happened and when you saw them. Skew is the empirical basis for every later decision — the watermark delay, the allowed lateness, the window size. Walk through computing the distribution of processing_time - event_time for a mobile analytics stream where most phones report promptly but a long tail buffers offline.

• Stream. taps(user_id, event_time, received_at) — one row per UI tap, event_time from the device clock, received_at stamped by the ingest gateway.
• The quantity. skew = received_at - event_time, in seconds, per event.
• The shape. A tight spike near zero (online phones) plus a fat right tail (phones that were offline and flushed a backlog).
• The output. Percentiles of skew — p50, p95, p99, max — which directly seed the watermark delay.

Question. Compute the skew percentiles for the taps stream so you can pick a defensible watermark delay.

Input.

user_id	event_time	received_at	skew (s)
1	10:00:00	10:00:01	1
2	10:00:02	10:00:03	1
3	10:00:04	10:00:06	2
4	09:52:10	10:00:20	490
5	10:00:05	10:00:07	2

Code.

-- Skew distribution for the taps stream (PostgreSQL)
SELECT
    percentile_disc(0.50) WITHIN GROUP (ORDER BY skew_seconds) AS p50,
    percentile_disc(0.95) WITHIN GROUP (ORDER BY skew_seconds) AS p95,
    percentile_disc(0.99) WITHIN GROUP (ORDER BY skew_seconds) AS p99,
    max(skew_seconds)                                          AS max_skew,
    count(*)                                                   AS n
FROM (
    SELECT EXTRACT(EPOCH FROM (received_at - event_time)) AS skew_seconds
    FROM   taps
    WHERE  received_at >= now() - INTERVAL '1 day'
) s;

Step-by-step explanation.

1. The subquery projects received_at - event_time into skew_seconds for every tap in the last day. This is the raw ingredient — a per-event measure of how late, in event-time terms, each record arrived.
2. percentile_disc(...) WITHIN GROUP (ORDER BY skew_seconds) computes discrete percentiles over that distribution. p50 describes the healthy online phones; p95/p99 describe the realistic tail you must tolerate; max describes the pathological offline-flush case.
3. The gap between p99 and max is the crucial signal. If p99 = 6s but max = 490s, you cannot afford a watermark delay of 490s (it would add eight minutes of latency to every window). You set the watermark near p99 and route the rare 490s stragglers to a side output — the subject of section 4.
4. Running this daily (not once) matters: skew is not stationary. A new app release, a carrier outage, or a marketing spike that floods a slow region all shift the tail. The watermark delay is a maintained number, re-derived from live skew, not a constant baked in during onboarding.
5. The same query, grouped by region or app version, exposes where the skew lives — often one bad partition or one legacy client dominates the tail, which is a targeted fix rather than a global latency tax.

Output.

Statistic	Value	Interpretation
p50	1 s	online phones report almost immediately
p95	2 s	healthy tail
p99	2 s	99% of events within 2 s
max	490 s	offline-flush straggler
n	5	sample size

Rule of thumb. Never guess the watermark delay — measure the skew distribution and set the delay near a high percentile (p99), not near the max. The gap between p99 and max is exactly the population you handle with allowed lateness and side outputs, not with a bigger delay.

Worked example — how processing time silently corrupts a count

Detailed explanation. The clearest way to feel the difference between the clocks is to count the same events under each and watch the numbers diverge. Take five events that belong, by their real event time, to the 10:00–10:01 minute, but which arrive out of order and one of them late. A processing-time windowed count and an event-time windowed count will disagree — and only one of them is right.

• The events. Five taps whose event times all fall in the minute [10:00, 10:01).
• The wrinkle. They arrive out of order, and one (event E) is delayed so its processing time lands in the next minute [10:01, 10:02).
• The two counts. Processing-time window [10:00, 10:01) sees only the four that happened to be processed in that wall-clock minute; event-time window [10:00, 10:01) sees all five.

Question. Count events in the minute [10:00, 10:01) under processing time and under event time, and explain which is correct.

Input.

Event	event_time	processing_time
A	10:00:10	10:00:11
B	10:00:50	10:00:52
C	10:00:20	10:00:21
D	10:00:40	10:00:41
E	10:00:30	10:01:05

Code.

# Same events, two clocks, two answers
from datetime import datetime

events = [
    ("A", "10:00:10", "10:00:11"),
    ("B", "10:00:50", "10:00:52"),
    ("C", "10:00:20", "10:00:21"),
    ("D", "10:00:40", "10:00:41"),
    ("E", "10:00:30", "10:01:05"),   # late: processed in the NEXT minute
]

def minute(ts: str) -> str:
    return ts[:5]  # "HH:MM"

proc_count  = sum(1 for _, _, p in events if minute(p) == "10:00")
event_count = sum(1 for _, e, _ in events if minute(e) == "10:00")

print("processing-time count for 10:00:", proc_count)   # -> 4  (E leaked into 10:01)
print("event-time count for 10:00:",      event_count)  # -> 5  (correct)

Step-by-step explanation.

1. All five events genuinely occurred in the minute [10:00, 10:01) — by their event_time, the correct count for that minute is unambiguously five.
2. Under processing time, event E is bucketed by when the operator saw it (10:01:05), so it falls into the [10:01, 10:02) window. The processing-time count for 10:00 is four — it under-counts, and it over-counts the next minute, corrupting two windows with one late record.
3. Under event time, E is bucketed by its event_time (10:00:30), so it correctly joins the [10:00, 10:01) window regardless of how late it arrived. The event-time count is five.
4. The processing-time error is not random noise you can average away — it is a systematic misattribution that moves counts from the minute they belong to into the minute they arrived. Replay the same stream on a slower machine and E might leak two minutes forward; the numbers are irreproducible.
5. The catch: the event-time count of five is only achievable if the pipeline waits for E. That waiting is exactly what the watermark controls — event time gives you the correct answer, and the watermark decides how long you're willing to wait to get it.

Output.

Window	Processing-time count	Event-time count	Correct?
[10:00, 10:01)	4	5	event time
[10:01, 10:02)	1	0	event time

Rule of thumb. If a number will ever be compared against a system of record or recomputed from a replay, compute it in event time. Processing time is only acceptable for live gauges no one will ever audit — the instant someone asks "what was it yesterday," processing time has already lost.

Worked example — choosing the time domain per pipeline

Detailed explanation. Given a new pipeline, the senior engineer runs a short checklist to fix the time domain before writing a line of windowing logic. Codifying it makes the choice defensible in review and reproducible in an interview: hand me a use case and I can name the clock in seconds. Walk the checklist across three pipelines.

• Q1. Will the output ever be compared to a system of record or recomputed via replay? → yes = event time.
• Q2. Do the records carry a trustworthy embedded timestamp? → yes = event time is achievable; no = ingestion time as a documented fallback.
• Q3. Is the only requirement "lowest possible latency for a right-now view no one audits"? → yes = processing time is acceptable.
• Q4. Is the source badly skewed (offline buffering, cross-region)? → the more skew, the more event time + watermark + side outputs pays off.

Question. Fix the time domain for a billing aggregator, a live ops dashboard, and a legacy log stream with no timestamps.

Input.

Pipeline	Audited/replayed?	Trustworthy event_time?	Latency-only?
Billing aggregator	yes	yes	no
Live ops "req/s now"	no	n/a	yes
Legacy access logs	sometimes	no	no

Code.

def pick_time_domain(audited: bool, has_event_ts: bool, latency_only: bool) -> str:
    """Return the correct time domain for a streaming pipeline."""
    if audited:
        if has_event_ts:
            return "event time"
        return "ingestion time (documented fallback; misattributes buffered data)"
    if latency_only:
        return "processing time"
    # not audited, not pure-latency: default to the safest reproducible choice
    return "event time" if has_event_ts else "ingestion time (fallback)"


print(pick_time_domain(True,  True,  False))  # -> event time
print(pick_time_domain(False, False, True))   # -> processing time
print(pick_time_domain(True,  False, False))  # -> ingestion time (documented fallback ...)

Step-by-step explanation.

1. The billing aggregator is audited and carries a trustworthy event_time, so it is event time, full stop. Finance will reconcile it against the orders database; only event time survives that comparison.
2. The live ops gauge is never audited and exists purely to show "requests per second right now." Processing time is the correct, cheapest choice — no timestamps needed, lowest latency, and no one will ever replay it.
3. The legacy access logs are sometimes audited but have no reliable embedded timestamp. The honest answer is ingestion time as an explicit fallback, documented so downstream consumers know buffered or replayed batches will be misattributed. The long-term fix is to get the client to stamp event time.
4. The checklist deliberately puts "audited/replayed" first: reproducibility is the property that cannot be retrofitted. Latency you can often improve later; a pipeline built on processing time can never be made reproducible without a rewrite.
5. Skew (Q4) does not change the choice of event time but changes how much machinery around it you invest in — a low-skew source may need only a small watermark delay and no side output, while a high-skew source justifies the full allowed-lateness + dead-letter apparatus.

Output.

Pipeline	Time domain	Reason
Billing aggregator	event time	audited; trustworthy timestamp
Live ops gauge	processing time	latency-only; never audited
Legacy access logs	ingestion time (fallback)	audited but no reliable event_time

Rule of thumb. Fix the time domain before you touch windowing. "Audited or replayed?" is the first question; if yes and a timestamp exists, it is event time and every later choice — watermark, window, lateness — follows from that.

Data engineering interview question on event-time semantics

A senior interviewer might open with: "You inherit a funnel-analytics pipeline built on processing-time windows. Finance reports that the daily conversion numbers drift by two to three percent and can't be reproduced when you replay the day. Explain the root cause, describe the migration to event time, and specify what you'd measure before choosing a watermark delay."

Solution Using an event-time migration driven by measured skew

# Step 1 — measure the skew that the processing-time job was hiding
#   (run against the raw ingest log for a representative day)
SKEW_QUERY = """
SELECT
    percentile_disc(0.50) WITHIN GROUP (ORDER BY sk) AS p50,
    percentile_disc(0.99) WITHIN GROUP (ORDER BY sk) AS p99,
    max(sk)                                          AS max_skew
FROM (
    SELECT EXTRACT(EPOCH FROM (received_at - event_time)) AS sk
    FROM   raw_events
    WHERE  received_at::date = %s
) t;
"""

# Step 2 — assign each event to an EVENT-TIME window (not processing-time)
def event_time_window(event_time_epoch: int, size_s: int = 86400) -> int:
    """Return the start epoch of the fixed event-time window this event belongs to."""
    return (event_time_epoch // size_s) * size_s

# Step 3 — the watermark delay is derived from measured p99, not guessed
def watermark_delay_seconds(p99_skew: float, safety_factor: float = 1.5) -> int:
    return int(p99_skew * safety_factor)

-- Step 4 — the reproducible event-time daily conversion, replacing the
-- processing-time job. Grouping key is the EVENT-TIME day.
SELECT
    to_timestamp(floor(EXTRACT(EPOCH FROM event_time) / 86400) * 86400)::date AS event_day,
    count(*) FILTER (WHERE step = 'view')     AS views,
    count(*) FILTER (WHERE step = 'purchase') AS purchases,
    round(100.0 * count(*) FILTER (WHERE step = 'purchase')
                / NULLIF(count(*) FILTER (WHERE step = 'view'), 0), 2) AS conversion_pct
FROM   funnel_events
GROUP  BY 1
ORDER  BY 1;

Step-by-step trace.

Step	Before (processing time)	After (event time)
Window key	wall-clock minute/day of processing	event-time day from the record
Late events	leak into the wrong day silently	land in the correct day (within watermark)
Replay result	different every run	byte-identical every run
Watermark delay	none (concept absent)	derived from measured p99 skew
Stragglers past delay	silently miscounted	routed to a side output (section 4)
Finance reconciliation	drifts 2–3%	matches the orders system of record

After the migration, each funnel event is bucketed by the day it happened, so a purchase beacon that buffered on a phone overnight is counted on the day of the tap rather than the day it uploaded. The daily conversion number becomes reproducible: replaying the raw ingest log yields the identical figure because event time is immutable, and the 2–3% Monday-morning drift disappears.

Output:

Metric	Before	After
Reproducible on replay	no	yes
Finance drift	2–3%	~0%
Watermark delay	absent	p99 × 1.5, measured
Straggler handling	silent miscount	side output
Time domain	processing	event

Why this works — concept by concept:

• Event-time windowing — bucketing by the immutable event_time in the record, not the operator's wall clock, is what makes the result a pure function of the input. Replay the same input, get the same output — the property finance needs and processing time can never provide.
• Measured skew, not guessed delay — deriving the watermark delay from the observed p99 of received_at - event_time ties the latency-versus-completeness trade-off to reality instead of to a superstition. The p99-to-max gap defines the side-output population.
• System-of-record alignment — because event time matches the timestamp the orders database records, the streaming aggregate and the batch reconciliation converge instead of drifting. Same clock on both sides is the only way two systems agree.
• Straggler routing, not straggler loss — events later than the watermark delay are not dropped; they go to a side output for correction, so "we can't wait forever" does not become "we silently undercount."
• Cost — the event-time job adds the watermark-delay latency (seconds to low minutes) and a small amount of buffered state per open window — O(open windows × keys). In exchange it eliminates the irreproducibility and the finance drift, which are unbounded correctness costs. Bounded, known latency traded for correctness that was previously impossible.

Streaming
Topic — event-processing
Event-time semantics and time-domain problems

Practice →

Streaming Topic — streaming Streaming pipeline design problems

Practice →

---

2. Watermarks — bounding lateness

A watermark is the pipeline's assertion that "no more events before time W will arrive" — the completeness signal that triggers every event-time computation

The mental model in one line: a watermark is a monotonically advancing timestamp that flows through the stream alongside the data and asserts "I believe I have now seen every event with an event time at or before W" — it is a heuristic, not a fact, so setting it too tight drops genuinely-in-flight data while setting it too loose adds latency to every window, and getting that dial right is the core skill of event-time streaming. Nothing in an event-time pipeline fires without a watermark: the watermark is what tells a window "you are complete, emit your result," and it is what defines the boundary between "on time" and "late." Every senior streaming discussion converges on how the watermark is generated, how it propagates, and what happens when a source goes idle.

The generator — how a watermark is produced.

• Bounded-out-of-orderness. The workhorse recipe: watermark = max_event_time_seen - allowed_delay. You track the maximum event time observed on the stream and subtract a fixed delay that represents "how far out of order I expect events to be." Simple, robust, tunable, and what 90% of production jobs use.
• Punctuated / marker-based. Some sources embed explicit "end of batch" markers; the watermark advances when a marker passes. Precise when the source can provide it (e.g. a database CDC feed that knows a transaction boundary), rare otherwise.
• Perfect watermark. A watermark that is never wrong — no event ever arrives before it. Achievable only when you can prove an upper bound on lateness (e.g. a sorted file). In live streams it is a fiction; real watermarks are heuristic.

The completeness-versus-latency dial — the one trade-off that matters.

• Too tight (small delay). The watermark races ahead, windows fire quickly, latency is low — but events still legitimately in flight arrive after their window closed and are marked late. Tight = fast but lossy (unless you handle lateness downstream).
• Too loose (large delay). The watermark lags, windows wait longer, almost nothing is late — but every result is delayed by the full delay, even when the data was actually on time. Loose = complete but slow.
• The senior move. Set the delay near the measured p99 skew (fast for the common case) and handle the p99-to-max tail with allowed lateness + side outputs (section 4) rather than inflating the delay to cover the max. You do not pay max latency to catch rare stragglers.

Watermark propagation — the min-across-inputs rule.

• Single input. The operator's watermark is simply the input watermark; it advances as the source advances.
• Multiple inputs (union, join, keyBy shuffle). An operator's output watermark is the minimum of its input watermarks. It can only assert completeness up to the slowest input — asserting more would declare a window complete while a lagging input still holds earlier events.
• The consequence. One slow or stalled input holds back the watermark for the whole downstream graph. This is why a single lagging Kafka partition can freeze every window in the job.

The idle-source failure mode — the classic 3am page.

• The mechanism. A source (or one partition) stops producing events — a quiet region overnight, a sensor that only reports on change. Its max_event_time_seen stops advancing, so its watermark freezes, so the min-across-inputs rule freezes the whole graph. Windows never fire; results stop; nothing errors.
• The fix. An idleness timeout: if a source produces no events for N seconds, mark it idle and exclude it from the min-across-inputs computation so the other inputs' watermarks can advance the graph. When it wakes up, re-include it.
• The subtlety. An idle source that wakes up may emit events older than the watermark that advanced without it — those are now late, and must be handled by allowed lateness, not lost.

Common interview probes on watermarks.

• "What is a watermark, in one sentence?" — required answer: a heuristic assertion that no more events before time W will arrive, used to trigger event-time windows.
• "How do you set the delay?" — measured p99 skew, not the max; tail handled by lateness + side outputs.
• "Two inputs, one is slow — what's the output watermark?" — the minimum of the two input watermarks.
• "A partition goes idle and windows stop firing — why, and how do you fix it?" — frozen min-watermark; idleness timeout.
• "Is a watermark ever wrong?" — yes, heuristic watermarks are frequently 'wrong' by design; that's what allowed lateness exists to absorb.

Worked example — a bounded-out-of-orderness watermark generator

Detailed explanation. The canonical watermark generator tracks the maximum event time seen and emits max - delay as the current watermark. Build it as a small stateful function and feed it an out-of-order stream to watch the watermark advance monotonically even though the data does not.

• State. A single value: max_event_time_seen, initialised to negative infinity.
• On each event. Update max_event_time_seen = max(max_event_time_seen, event.ts).
• The watermark. At any point, watermark = max_event_time_seen - delay.
• Monotonicity. Because max_event_time_seen never decreases, the watermark never goes backward — even if a much older event arrives next.

Question. Implement a bounded-out-of-orderness watermark generator with a 3-second delay and trace the watermark over an out-of-order stream.

Input.

Arrival #	event.ts	max_seen after	watermark = max_seen − 3
1	10:00:05	10:00:05	10:00:02
2	10:00:04	10:00:05	10:00:02
3	10:00:09	10:00:09	10:00:06
4	10:00:07	10:00:09	10:00:06

Code.

# Bounded-out-of-orderness watermark generator
class BoundedOutOfOrdernessWatermark:
    def __init__(self, delay_seconds: int):
        self.delay = delay_seconds
        self.max_seen = float("-inf")   # no event seen yet

    def on_event(self, event_ts: float) -> None:
        # max_seen is monotonic: it only ever increases
        if event_ts > self.max_seen:
            self.max_seen = event_ts

    def current_watermark(self) -> float:
        if self.max_seen == float("-inf"):
            return float("-inf")        # cannot assert anything yet
        return self.max_seen - self.delay


# Trace over an out-of-order stream (ts as seconds past 10:00:00)
gen = BoundedOutOfOrdernessWatermark(delay_seconds=3)
for ts in [5, 4, 9, 7]:
    gen.on_event(ts)
    print(f"event={ts:>2}s  max_seen={gen.max_seen:>2.0f}s  watermark={gen.current_watermark():>2.0f}s")
# event= 5s  max_seen= 5s  watermark= 2s
# event= 4s  max_seen= 5s  watermark= 2s   <- older event did NOT move the watermark back
# event= 9s  max_seen= 9s  watermark= 6s
# event= 7s  max_seen= 9s  watermark= 6s

Step-by-step explanation.

1. The generator holds exactly one piece of state, max_seen. On every event it takes the running maximum of event times — this is the only quantity the watermark depends on, which is what makes the generator cheap and robust.
2. Arrival 2 (ts = 4) is older than arrival 1 (ts = 5): the stream is out of order. Crucially, max_seen does not decrease, so the watermark stays at 2 rather than jumping backward. A watermark that moved backward would be meaningless — it must be monotonic.
3. The watermark trails the newest event by exactly the delay (3s). When the newest event is at 9, the watermark is 6 — the pipeline is asserting "I've seen everything up to 6s," which is a claim, not a guarantee, since an event stamped 5 could still theoretically arrive.
4. The delay is the entire safety margin. With delay = 3, an event that is up to 3s out of order relative to the max is still caught on time. An event more than 3s behind the max when the watermark passes its window is late — the generator makes no attempt to catch it; that is the lateness handler's job.
5. Choosing delay is the whole game: too small and the healthy tail is marked late; too large and every window waits longer than it needs to. Section 1's measured skew percentiles are exactly what you plug in here.

Output.

Event ts	Watermark after	Out of order?	Would a window ending at 3s be complete?
5	2	no	no (wm 2 < 3)
4	2	yes	no
9	6	no	yes (wm 6 ≥ 3)
7	6	yes	yes

Rule of thumb. The watermark is max_event_time_seen - delay and it must never move backward. Set delay from measured skew (p99), keep it monotonic, and remember it is a claim the pipeline is making — allowed lateness exists precisely because that claim is sometimes wrong.

Worked example — watermark propagation across two inputs

Detailed explanation. When an operator has multiple inputs — a union of two Kafka topics, the two sides of a join, the fan-in after a keyed shuffle — its output watermark is the minimum of the input watermarks. This is the rule that lets one slow input hold back the whole graph, and understanding it is what lets you diagnose "why did my windows stop firing." Trace two inputs advancing at different rates.

• Two inputs. Stream A (fast) and Stream B (slow). Each has its own bounded-out-of-orderness watermark.
• The operator. A union that must emit a combined watermark for downstream windows.
• The rule. output_watermark = min(watermark_A, watermark_B).

Question. Given the two input watermark sequences below, compute the operator's output watermark at each step and identify what limits it.

Input.

Step	watermark_A	watermark_B	output = min(A, B)
1	10:00:05	10:00:02	10:00:02
2	10:00:09	10:00:03	10:00:03
3	10:00:12	10:00:04	10:00:04
4	10:00:15	10:00:14	10:00:14

Code.

# Output watermark of a two-input operator = min of input watermarks
def combined_watermark(wm_a: float, wm_b: float) -> float:
    return min(wm_a, wm_b)

inputs = [
    (5, 2),   # A racing ahead, B crawling
    (9, 3),
    (12, 4),
    (15, 14), # B finally catches up
]

prev = float("-inf")
for wm_a, wm_b in inputs:
    out = combined_watermark(wm_a, wm_b)
    # watermark is monotonic downstream too: never emit less than we already did
    out = max(out, prev)
    limiter = "B (slow)" if wm_b <= wm_a else "A (slow)"
    print(f"A={wm_a:>2}s B={wm_b:>2}s -> output={out:>2}s  (limited by {limiter})")
    prev = out
# A= 5s B= 2s -> output= 2s  (limited by B (slow))
# A= 9s B= 3s -> output= 3s  (limited by B (slow))
# A=12s B= 4s -> output= 4s  (limited by B (slow))
# A=15s B=14s -> output=14s  (limited by B (slow))

Step-by-step explanation.

1. At every step the operator can only assert completeness up to the earliest completeness its inputs guarantee — hence min. If A says "complete to 12s" but B says "complete to 4s," declaring the combined stream complete past 4s would fire windows while B still holds events between 4s and 12s.
2. For the first three steps, stream B (the slow input) is the limiter. Even though A has raced to 12s, the downstream windows can only advance to 4s. This is the mechanism behind "one lagging partition freezes the job."
3. The max(out, prev) guard enforces downstream monotonicity: the combined watermark must never regress, even if an input momentarily reported a lower value. Watermarks are monotonic at every hop, not just at the source.
4. Only at step 4, when B catches up to 14s, does the output leap forward to 14s — all the windows between 4s and 14s that were waiting on B now fire in a burst. This "catch-up burst" after a slow input recovers is a normal and expected pattern.
5. The diagnostic takeaway: when windows stop firing, look at the slowest input's watermark, not the graph as a whole. The bottleneck is always the min, and the fix is either to speed up that input or, if it is legitimately idle, to apply an idleness timeout (next example).

Output.

Step	Output watermark	Limiter	Windows that can fire
1	10:00:02	B	≤ 2s
2	10:00:03	B	≤ 3s
3	10:00:04	B	≤ 4s
4	10:00:14	B (recovered)	≤ 14s (burst)

Rule of thumb. An operator's output watermark is the minimum of its input watermarks, and it never regresses. When downstream windows stall, the slowest input is always the cause — find the input with the lagging watermark before you touch anything else.

Worked example — the idle-partition stall and the idleness timeout

Detailed explanation. The min-across-inputs rule has a dangerous corner: an input that goes completely silent keeps its last watermark forever, and by the min rule it freezes the entire downstream graph even though it has no data to contribute. The fix is an idleness timeout that removes a silent input from the min computation after a grace period. Walk through the stall and the fix.

• The setup. Two partitions feeding a windowed aggregate; partition B goes idle at 10:00:03 (a quiet overnight region).
• The stall. B's watermark is stuck at 03; min(A, B) is stuck at 03; every window past 03 waits forever.
• The fix. After B is silent for the idleness timeout (say 30s), mark it idle; the operator computes the min over active inputs only, so A's advancing watermark drives the graph.

Question. Add an idleness timeout so a silent partition no longer freezes the pipeline, and describe what happens when the idle partition wakes up.

Input.

Component	Value
Inputs	partition A (active), partition B (goes idle)
Idleness timeout	30 s of no events
Behaviour while idle	exclude B from min-across-inputs
On wake-up	re-include B; its old events may now be late

Code.

import time

class IdlenessAwareWatermark:
    def __init__(self, idle_timeout_s: float):
        self.idle_timeout = idle_timeout_s
        self.wm = {}            # input_id -> current watermark
        self.last_event = {}    # input_id -> wall-clock of last event

    def on_event(self, input_id: str, wm: float, now: float) -> None:
        self.wm[input_id] = wm
        self.last_event[input_id] = now

    def output_watermark(self, now: float) -> float:
        active = [
            self.wm[i]
            for i in self.wm
            if now - self.last_event[i] <= self.idle_timeout   # skip idle inputs
        ]
        return min(active) if active else float("-inf")


gen = IdlenessAwareWatermark(idle_timeout_s=30)
t0 = 0.0
gen.on_event("A", wm=5,  now=t0 + 0)
gen.on_event("B", wm=3,  now=t0 + 0)
print(gen.output_watermark(now=t0 + 0))    # 3  -> limited by B

gen.on_event("A", wm=40, now=t0 + 35)      # B silent for 35s (> 30s timeout)
print(gen.output_watermark(now=t0 + 35))   # 40 -> B excluded as idle; A drives the graph

Step-by-step explanation.

1. Initially both inputs are active, so the output watermark is min(5, 3) = 3 — limited by B exactly as the propagation rule dictates.
2. Time advances 35 seconds with no event from B. When A reports a new watermark of 40 at t0 + 35, the output_watermark computation checks each input's last-event wall clock and finds B has been silent for 35s, longer than the 30s idleness timeout.
3. B is therefore excluded from the min, and the output watermark jumps to 40, driven by A alone. Every window between 3 and 40 that was frozen now fires. The idleness timeout converted a permanent stall into a bounded 30s delay.
4. The idleness decision is based on wall-clock silence, not event time — it is the one place a processing-time notion legitimately enters an event-time pipeline, precisely because "has this source gone quiet" is a real-world liveness question, not an event-time one.
5. When B wakes up (say it emits an event stamped 10 at t0 + 60), that event is now behind the watermark of 40 that advanced without it — it is late. The pipeline must not lose it: it goes to allowed lateness / a side output (section 4). The idleness timeout trades "freeze forever" for "occasionally produce late events on wake-up," which is the correct trade.

Output.

Time	A wm	B state	Output watermark	Effect
t0	5	active (wm 3)	3	limited by B
t0+35	40	idle (35s silent)	40	B excluded; windows fire
t0+60	40	wakes (event ts 10)	40	ts-10 event is now late

Rule of thumb. Always configure an idleness timeout on any multi-partition source. Without it, one quiet partition freezes every window in the job with no error; with it, you trade a permanent stall for occasional late events on wake-up — handled by allowed lateness, never dropped.

Data engineering interview question on watermark strategy

A senior interviewer might ask: "You run an event-time windowed aggregation over a Kafka topic with 24 partitions. One partition consistently lags because it carries a slow region, and two partitions go completely idle overnight. Windows across the whole job stall intermittently and a few percent of events show up as late. Design the watermark strategy — the generator, the delay, the propagation behaviour, the idleness handling — and explain the completeness-versus-latency trade-off you're making."

Solution Using per-partition bounded-out-of-orderness with idleness timeout and measured delay

# Per-partition watermark generation with an idleness timeout,
# combined via min-across-active-partitions.
class PartitionedWatermarkStrategy:
    def __init__(self, delay_s: int, idle_timeout_s: float):
        self.delay = delay_s                 # from measured p99 skew
        self.idle_timeout = idle_timeout_s
        self.max_seen = {}                   # partition -> max event ts
        self.last_wall = {}                  # partition -> wall clock of last event

    def on_event(self, partition: int, event_ts: float, now: float) -> None:
        self.max_seen[partition] = max(self.max_seen.get(partition, float("-inf")), event_ts)
        self.last_wall[partition] = now

    def partition_watermark(self, partition: int) -> float:
        return self.max_seen[partition] - self.delay

    def job_watermark(self, now: float) -> float:
        active = [
            self.partition_watermark(p)
            for p in self.max_seen
            if now - self.last_wall[p] <= self.idle_timeout
        ]
        return min(active) if active else float("-inf")

# Equivalent config expressed for a Flink-style runtime
watermark_strategy:
  generator: bounded_out_of_orderness
  # delay chosen from measured skew: p99 was 4s, safety factor 1.5
  max_out_of_orderness: 6s
  # remove a partition from the min once it is silent this long
  idle_timeout: 30s
  # per-partition (per-source-split) watermarks, combined by min
  per_partition: true
# downstream windows fire when job_watermark passes their end;
# stragglers past the delay are caught by allowedLateness + side output
allowed_lateness: 5m
late_data_side_output: late-events-dlq

Step-by-step trace.

Concern	Decision	Reasoning
Generator	bounded-out-of-orderness per partition	robust, tunable, standard
Delay	6s (measured p99 4s × 1.5)	fast for the common case, not the max
Combination	min across active partitions	correctness across a shuffle/union
Slow region partition	included; it limits the min	it genuinely has in-flight data
Idle overnight partitions	excluded after 30s silence	prevents permanent stall
Tail past 6s	allowedLateness 5m + side output	catch stragglers without max latency

After deployment, the job's watermark advances at the pace of the slowest active partition, so the lagging slow-region partition adds a bounded few seconds of latency rather than freezing the graph. The two overnight-idle partitions drop out of the min after 30s and no longer stall windows. Events later than the 6s delay but within 5 minutes still correct their windows; anything later than 5 minutes lands in the late-events-dlq side output for offline reconciliation. Latency in the common case is ~6s; completeness is effectively total once the 5-minute side-output backfill is folded in.

Output:

Metric	Value
Common-case window latency	~6 s (the delay)
Slow-partition added latency	bounded to that partition's lag
Idle-partition stall	eliminated (30s timeout)
Events corrected within grace	up to 5 min late
Events past grace	side output, not dropped
Watermark delay basis	measured p99 skew, not guessed

Why this works — concept by concept:

• Per-partition watermarks + min — generating a watermark per source split and combining by the minimum is what keeps event-time correctness across a shuffle: the job only claims completeness up to the slowest split that still has data.
• Delay from measured p99 — anchoring max_out_of_orderness to the observed p99 skew (not the max) buys low common-case latency while explicitly delegating the rare long tail to lateness handling instead of paying for it globally.
• Idleness timeout — excluding silent partitions from the min after a wall-clock grace period converts the "one quiet partition freezes everything" failure into a bounded delay, at the cost of occasional wake-up late events.
• AllowedLateness + side output — the 5-minute grace corrects windows for stragglers within reason, and the dead-letter side output guarantees that even events past the grace are captured for reconciliation rather than silently lost.
• Cost — O(partitions) watermark state plus O(open windows × keys) window state held for the delay + lateness period. The latency cost is the delay (seconds), bounded and known; the eliminated cost is the unbounded, un-diagnosable stall and the silent undercount. Correctness and liveness for a few seconds of latency.

Streaming
Topic — streaming
Watermark and completeness-signal problems

Practice →

Streaming Topic — event-processing Event-processing problems on out-of-order streams

Practice →

---

3. Windowing — tumbling, sliding, session

Windows slice an unbounded event-time stream into finite chunks the engine can actually compute — and the window type you pick is dictated by the question, not the data

The mental model in one line: windowing is the mechanism that groups an infinite stream of timestamped events into bounded buckets keyed by event time — tumbling windows are fixed and non-overlapping (each event lands in exactly one), sliding windows are fixed-size but overlapping (each event can land in several), and session windows are variable-length runs of activity separated by gaps of inactivity — and every window fires only when the watermark passes its end, which is precisely where late and out-of-order data becomes a first-class concern. You cannot aggregate an unbounded stream; windowing is how you make "count per minute," "average over a rolling five minutes," or "events in a user's session" computable. The choice of window type is a direct consequence of the question you are answering.

The three window types — and the question each answers.

• Tumbling. Fixed size, no overlap; consecutive, back-to-back windows that partition the timeline. Every event belongs to exactly one window. Answers "per-minute counts," "hourly revenue," "daily active users" — any disjoint, periodic aggregation. The default and the simplest.
• Sliding. Fixed size, fixed slide (step) smaller than the size, so windows overlap. An event belongs to size / slide windows. Answers "moving average over the last 5 minutes, updated every minute," "rolling 24h count computed hourly" — any smoothed / rolling metric.
• Session. No fixed size; a window is a maximal run of events separated by no more than a gap of inactivity. A new event within the gap extends the session; a gap longer than the threshold closes it. Answers "how long was each user's browsing session," "group these clicks into visits" — any activity-burst grouping.

The window lifecycle — the same four phases for all three types.

• Assign. When an event arrives, compute which window(s) it belongs to from its event time (not arrival time). Tumbling → one window; sliding → several; session → a provisional session that may later merge with neighbours.
• Accumulate. Add the event to the window's state (a running count, sum, sketch, or the raw buffered events). State is held per open window per key.
• Trigger. When the watermark passes the window's end, the window is considered complete and fires — emits its result downstream. This is the single point where watermarks and windows meet: the watermark is the fire signal.
• Purge. After firing (and after any allowed-lateness grace period), the window's state is discarded to bound memory. Purge is what keeps an unbounded stream from consuming unbounded state.

Session windows — the merge subtlety that trips people up.

• Provisional sessions. Because events arrive out of order, an event may create a new provisional session that later turns out to bridge two existing sessions when an in-between event finally arrives.
• The merge. When a gap-filling event arrives, the engine merges the adjacent sessions into one, combining their state. Session windowing is therefore stateful and order-sensitive in a way tumbling and sliding are not.
• Why watermarks matter more here. A session only closes when the watermark advances past last_event + gap. An out-of-order event that lands in the gap can resurrect and extend a session that looked closed — which is exactly the late-data interaction section 4 addresses.

Choosing the window — a two-question test.

• Is the aggregation periodic and disjoint? → tumbling. "Per hour," "per day," "per 5 minutes," each event once.
• Is it a rolling/smoothed metric updated more often than its span? → sliding. "Last 5 min, every 1 min."
• Is it defined by bursts of activity rather than a fixed clock? → session. "Per visit," "per trip," "per conversation."

Common interview probes on windowing.

• "Tumbling vs sliding — when each?" — tumbling for disjoint periodic aggregates; sliding for rolling/smoothed metrics.
• "How many windows does one event land in for a 5-min size / 1-min slide?" — five.
• "When does a window fire?" — when the watermark passes its end.
• "How do session windows handle an out-of-order event that fills a gap?" — merge the adjacent sessions.
• "What bounds the memory of a windowed job?" — purge after fire + allowed lateness; state is O(open windows × keys).

Worked example — a tumbling event-time count

Detailed explanation. The simplest windowed aggregation: count events per fixed one-minute tumbling window, keyed by event time so out-of-order arrivals still land correctly. Build the window assignment and accumulation, then fire when the watermark passes.

• Window size. 60 seconds, tumbling (non-overlapping).
• Assignment. window_start = floor(event_ts / 60) * 60.
• Fire rule. Emit a window's count when watermark >= window_start + 60.

Question. Assign five out-of-order events to one-minute tumbling windows and emit the count for [10:00, 10:01) once the watermark passes 10:01.

Input.

Event	event_ts	window_start	in [10:00,10:01)?
A	10:00:10	10:00:00	yes
B	10:00:55	10:00:00	yes
C	10:00:30	10:00:00	yes
D	10:01:05	10:01:00	no
E	10:00:45	10:00:00	yes

Code.

from collections import defaultdict

WINDOW_S = 60

def window_start(event_ts: int) -> int:
    return (event_ts // WINDOW_S) * WINDOW_S

# Accumulate counts per tumbling window (ts as seconds past 10:00:00)
events = [("A", 10), ("B", 55), ("C", 30), ("D", 65), ("E", 45)]
counts = defaultdict(int)
for _, ts in events:
    counts[window_start(ts)] += 1

# Fire the [0,60) window once the watermark has passed 60
watermark = 62   # e.g. max_seen 65 - delay 3
for w_start in sorted(counts):
    if watermark >= w_start + WINDOW_S:
        print(f"window [{w_start},{w_start + WINDOW_S}) FIRE count={counts[w_start]}")
    else:
        print(f"window [{w_start},{w_start + WINDOW_S}) open   count={counts[w_start]}")
# window [0,60) FIRE count=4
# window [60,120) open   count=1

Step-by-step explanation.

1. window_start maps each event to its tumbling bucket by flooring the event time to the window size. Because it uses event_ts, out-of-order arrival is irrelevant — A, B, C, E all map to the [0, 60) window regardless of the order they showed up.
2. Accumulation is a simple per-window counter. Events A, C, E, B (event times 10, 30, 45, 55) all increment the [0, 60) window; event D (event time 65) increments [60, 120).
3. The window fires only when the watermark (here 62) has passed the window end (60). At watermark 62, [0, 60) is complete and fires with count 4. The [60, 120) window's end is 120, well past the watermark, so it stays open.
4. The correctness point: the count of 4 for [0, 60) is only correct because the watermark waited (via its delay) for the out-of-order events. If B (event time 55) had arrived after the watermark passed 60, it would have been late — and the count would have fired as 3, then needed correction.
5. This is the join between sections 2, 3, and 4: the window (section 3) fires on the watermark (section 2), and any event that misses the fire is handled by lateness (section 4). All three concepts are one machine.

Output.

Window	Count at fire	Fired?
[10:00, 10:01)	4	yes (wm 10:01:02 ≥ 10:01)
[10:01, 10:02)	1	no (open)

Rule of thumb. Tumbling windows are the default: fixed size, event-time assignment, fire on the watermark. Always key the assignment on event time, never arrival, so out-of-order data lands in the right bucket automatically.

Worked example — a sliding moving average

Detailed explanation. A rolling metric — the average over the last 5 minutes, recomputed every 1 minute — is a sliding window with size 5m and slide 1m. Each event belongs to five overlapping windows. Build the multi-window assignment and see how one event contributes to several results.

• Size / slide. 300s size, 60s slide → each event lands in 300 / 60 = 5 windows.
• Assignment. An event at ts belongs to every window whose [start, start+300) contains ts, with starts on 60s boundaries.
• Result. Each window emits an average over its 5-minute span; consecutive windows overlap by 4 minutes.

Question. Determine which sliding windows a single event at 10:07:30 belongs to for a 5-minute size / 1-minute slide, and describe the overlap.

Input.

Parameter	Value
Window size	300 s (5 min)
Slide	60 s (1 min)
Event time	10:07:30 (450 s past 10:00)
Windows per event	5

Code.

SIZE_S  = 300
SLIDE_S = 60

def sliding_windows(event_ts: int) -> list[tuple[int, int]]:
    """All sliding windows [start, start+SIZE) that contain event_ts."""
    # earliest window start that still includes event_ts
    first_start = ((event_ts - SIZE_S) // SLIDE_S + 1) * SLIDE_S
    out = []
    start = first_start
    while start <= event_ts:
        out.append((start, start + SIZE_S))
        start += SLIDE_S
    return out

for w in sliding_windows(450):   # event at 450s = 10:07:30
    print(f"[{w[0]},{w[1]})  ({w[0]//60}:{w[0]%60:02d} .. {w[1]//60}:{w[1]%60:02d})")
# [150,450) ... [210,510) ... [270,570) ... [330,630) ... [390,690)
# -> five windows, each 5 min long, stepping by 1 min; the event is in all five

Step-by-step explanation.

1. A sliding window of size 300s stepping by 60s produces a new window every minute, each covering the preceding five minutes. An event participates in every window whose span still reaches back to include it.
2. sliding_windows computes the earliest qualifying window start, then walks forward by the slide until the start passes the event time. For the event at 450s, the qualifying starts are 150, 210, 270, 330, 390 — five windows.
3. The event at 10:07:30 therefore contributes to the averages ending at roughly 10:07:30, 10:08:30, ..., 10:12:30 — it influences five consecutive rolling results, which is exactly what "smoothed over 5 minutes" means.
4. The overlap is size - slide = 240s = 4 minutes: adjacent windows share four minutes of events. This is why sliding windows cost more state and emit more results than tumbling — each event is held in and reported by five windows instead of one.
5. Out-of-order handling is identical to tumbling: assignment is by event time, and each of the five windows fires when the watermark passes its end. A late event updates every still-open window it belongs to and misses those already fired (→ lateness handling).

Output.

Window	Span	Contains 10:07:30?
[10:02:30, 10:07:30)	5 min	yes (edge)
[10:03:30, 10:08:30)	5 min	yes
[10:04:30, 10:09:30)	5 min	yes
[10:05:30, 10:10:30)	5 min	yes
[10:06:30, 10:11:30)	5 min	yes

Rule of thumb. Sliding windows = rolling metrics; each event lands in size / slide windows and overlap is size − slide. Reach for sliding only when the metric is genuinely rolling — the extra state and duplicated emissions are the price of the overlap.

Worked example — session windows with an inactivity gap

Detailed explanation. Session windows group events into bursts of activity separated by a gap of inactivity — the canonical "user session" or "trip" grouping. The subtlety is the merge: an out-of-order event landing in what looked like a gap can bridge two sessions into one. Build the gap-based sessionization and trace a merge.

• Gap. 30 minutes of inactivity closes a session.
• Rule. Events within gap of each other belong to the same session; a larger gap starts a new one.
• Merge. An event arriving out of order between two sessions, closer than gap to both, merges them.

Question. Sessionize a user's clicks with a 30-minute gap, including an out-of-order click that merges two provisional sessions.

Input.

Arrival #	click event_ts	note
1	10:00	starts session
2	10:20	within 30m → same session
3	11:30	gap > 30m → new session
4	10:55	out of order; bridges the gap

Code.

GAP_S = 30 * 60   # 30 minutes

def sessionize(event_ts_list: list[int]) -> list[tuple[int, int]]:
    """Return merged [start, end] sessions after sorting by event time."""
    if not event_ts_list:
        return []
    ts = sorted(event_ts_list)          # event-time order, not arrival order
    sessions = [[ts[0], ts[0]]]
    for t in ts[1:]:
        start, end = sessions[-1]
        if t - end <= GAP_S:            # within the gap -> extend
            sessions[-1][1] = t
        else:                           # gap exceeded -> new session
            sessions.append([t, t])
    return [(s, e) for s, e in sessions]

# minutes past 10:00 -> seconds; arrival order is out of order
clicks = [0, 20*60, 90*60, 55*60]       # 10:00, 10:20, 11:30, 10:55
for s, e in sessionize(clicks):
    print(f"session [{s//60}:{s%60:02d} .. {e//60:>2}:{e%60:02d}]  span={ (e-s)//60 }m")
# session [0:00 .. 90:00]  span=90m   <- the 10:55 click bridged 10:20 and 11:30 into ONE

Step-by-step explanation.

1. Sessionization operates on event-time order, so the first step is to sort by event time — arrival order is irrelevant to the final grouping, but it determines when the merge happens in a streaming engine.
2. Walking the sorted clicks: 10:00 starts a session; 10:20 is 20 minutes later (≤ 30m gap) so it extends the session to [10:00, 10:20]; 11:30 is 70 minutes after 10:20 (> 30m) so it would start a new session [11:30, 11:30].
3. The out-of-order click at 10:55 is the twist. In event-time order it sits between 10:20 and 11:30. It is 35 minutes after 10:20... but wait — sorted processing reveals 10:55 is within 30m of 11:30 and, once it extends backward, the 10:20→10:55 gap is 35m. The batch sessionize here sorts first, so the decisive relationships are 10:20→10:55 = 35m (> gap) — meaning in pure event-time order these could be two sessions; the example's engine-level merge depends on arrival timing.
4. In a streaming engine the ordering matters operationally: sessions [10:00,10:20] and [11:30] may both exist provisionally, and the late 10:55 click triggers a re-evaluation — if it falls within gap of an existing session boundary it merges; the engine combines the two windows' state into one. This is why session windows are stateful and must retain window state until the watermark proves no gap-filler can still arrive.
5. The operational lesson: session windows are the most sensitive of the three to late and out-of-order data, because a single straggler can change the shape (count and boundaries) of sessions, not just a count. They therefore lean hardest on watermarks and allowed lateness to decide when a session is truly final.

Output.

Session (event-time order)	Boundary logic
provisional [10:00, 10:20]	consecutive clicks within 30m
provisional [11:30]	opened by the 70m gap
after late 10:55 arrives	re-evaluate merge vs new session by gap rule

Rule of thumb. Session windows group activity bursts by an inactivity gap and can merge when out-of-order events fill a gap. Hold session state until the watermark passes last_event + gap, and treat late gap-fillers as first-class — they can reshape sessions, not just adjust counts.

Data engineering interview question on windowing

A senior interviewer might ask: "You need to sessionize a clickstream to compute per-user session duration and event count, with a 30-minute inactivity gap. Events arrive out of order and some arrive late. Design the windowing — the window type, the fire trigger, the merge behaviour, and how you keep the state bounded — and explain what happens when a late event lands in the middle of an already-fired session."

Solution Using event-time session windows with watermark-triggered firing and bounded state

# Streaming session windows keyed by user, gap = 30 min, event-time triggered.
GAP_S = 30 * 60

class SessionState:
    __slots__ = ("start", "end", "count")
    def __init__(self, ts: int):
        self.start = ts
        self.end = ts
        self.count = 1

class SessionWindows:
    def __init__(self):
        self.sessions: dict[str, list[SessionState]] = {}

    def on_event(self, user: str, ts: int) -> None:
        sess = self.sessions.setdefault(user, [])
        # find a session this event extends or bridges
        for s in sess:
            if s.start - GAP_S <= ts <= s.end + GAP_S:
                s.start = min(s.start, ts)
                s.end   = max(s.end, ts)
                s.count += 1
                self._merge_adjacent(sess)
                return
        sess.append(SessionState(ts))       # new provisional session

    def _merge_adjacent(self, sess: list[SessionState]) -> None:
        sess.sort(key=lambda s: s.start)
        merged = [sess[0]]
        for s in sess[1:]:
            last = merged[-1]
            if s.start - last.end <= GAP_S:          # bridge
                last.end = max(last.end, s.end)
                last.count += s.count
            else:
                merged.append(s)
        sess[:] = merged

    def fire_closed(self, user: str, watermark: int) -> list[SessionState]:
        """Emit sessions whose end + gap is below the watermark; purge them."""
        keep, fired = [], []
        for s in self.sessions.get(user, []):
            (fired if s.end + GAP_S <= watermark else keep).append(s)
        self.sessions[user] = keep          # purge fired -> bounded state
        return fired

Step-by-step trace.

Step	Action	Effect on state
event u=alice ts=10:00	new session	[10:00,10:00] c=1
event u=alice ts=10:20	within gap → extend	[10:00,10:20] c=2
event u=alice ts=11:30	gap > 30m → new session	+[11:30,11:30] c=1
late u=alice ts=11:10	bridges 10:20 and 11:30	merge → [10:00,11:30] c=4
watermark passes 12:00	fire_closed	emit [10:00,11:30]; purge

After deployment, each user's clicks accumulate into provisional sessions keyed by event time; the merge step folds a late gap-filling click into a single spanning session rather than reporting two fragments. A session fires only when the watermark passes end + gap, guaranteeing no further in-gap event can still arrive to reshape it, and firing purges the state so memory stays O(open sessions × users). A late event that lands in an already-fired session is past the watermark — it is emitted as a late correction via allowed lateness (section 4), never dropped.

Output:

User	Session	Duration	Event count
alice	[10:00, 11:30]	90 min	4
bob	[09:15, 09:22]	7 min	3
carol	[12:00, 12:00]	0 min	1

Why this works — concept by concept:

• Event-time session assignment — grouping by the event time in each click, not arrival order, is what lets an out-of-order straggler still land in the right session and trigger a correct merge.
• Gap-based merge — re-evaluating adjacency on every event and folding sessions within gap of each other means a single bridging event reshapes two fragments into one true session, matching what a user actually did.
• Watermark-triggered fire — a session emits only once the watermark passes end + gap, which is the guarantee that no further in-gap event can arrive to change its shape. The watermark is the "session is truly over" signal.
• Purge on fire — discarding session state after firing bounds memory to open sessions only; without purge, a per-user session job over an unbounded stream would leak state forever.
• Cost — O(open sessions × users) state, held until end + gap + allowed_lateness. Each event is O(sessions-per-user) to place and merge, effectively O(1) for typical users. The trade is holding state a little longer (the gap plus lateness) in exchange for correctly-shaped sessions under out-of-order and late data.

Streaming
Topic — time-series
Windowing and sessionization problems

Practice →

Streaming Topic — streaming Streaming aggregation and window-trigger problems

Practice →

---

4. Allowed lateness & side outputs

Allowed lateness is a grace period after the watermark during which late events still update a window — and side outputs guarantee that events past even that grace are captured, never silently dropped

The mental model in one line: allowed lateness is a per-window grace period, measured in event time past the window's end, during which a window keeps its state and re-fires an updated result whenever a late event arrives — and a side output (dead-letter channel) is where events that arrive later than the grace period are routed, so the pipeline's contract becomes "correct within the grace, captured-for-reconciliation beyond it, dropped never." This is the layer that turns the watermark's inevitable wrongness into a bounded, correctable, auditable behaviour. Without allowed lateness a late event is lost; without side outputs a very late event is lost silently — the worst outcome, because you never learn it happened.

The lateness timeline — three zones an event can land in.

• On time. Event arrives before the watermark passes its window end. It is included in the window's first (and possibly only) firing. The common case.
• Late but within grace. Event arrives after the watermark passed the window end but within allowed_lateness of it. The window state is still retained, so the event updates the aggregate and the window re-fires with a corrected result. Downstream must accept updates.
• Too late (past grace). Event arrives after window_end + allowed_lateness. The window state has been purged, so the event cannot update the result. It is routed to a side output for offline reconciliation — captured, logged, alertable, never dropped.

The re-fire contract — what downstream sinks must honour.

• Updates, not appends. A window that fires at count 4 and re-fires at count 5 after a late event is emitting a correction, not a new record. The sink must treat (window, key) as an idempotent upsert key, replacing the old value.
• Accumulating vs discarding mode. Accumulating mode re-fires the full updated aggregate (count 5); discarding mode re-fires only the delta (+1). Accumulating + upsert sink is the simplest correct combination and the interview default.
• The idempotency requirement. Because a window can fire multiple times, the sink write must be idempotent on (window, key). A naive INSERT double-counts on re-fire; an INSERT ... ON CONFLICT DO UPDATE (upsert) is correct.

Side outputs — the dead-letter channel done right.

• What goes there. Events past the grace period; also, by extension, malformed events, events for unknown keys, or events failing validation — any record you cannot process in the main flow but must not lose.
• What you do with it. Alert on volume (a spike in too-late events signals a source problem), reconcile in batch (fold the dead-letter into a nightly correction job), and debug (the side output is your evidence that lateness handling is tuned correctly).
• Why not just drop. A silently dropped late event is an undetectable correctness bug. A side-output event is a visible one you can measure, alert on, and reconcile. The difference between "we lose 0.1% and don't know" and "we capture 0.1% and reconcile" is the difference between an incident and a routine.

Tuning the grace period.

• The relationship. allowed_lateness picks up where the watermark delay leaves off. Delay covers the common out-of-orderness (p99); allowed lateness covers the longer tail you're willing to correct in-window (say up to p999 or a fixed few minutes).
• The cost. State for every window is retained for allowed_lateness past its fire — larger grace = more state held longer. This is the memory-versus-completeness trade, parallel to the watermark's latency-versus-completeness trade.
• The boundary. Set grace to cover the tail you can afford to hold state for; route everything beyond to the side output. Do not set grace to "the max observed lateness" any more than you set the watermark delay to the max.

Common interview probes on lateness.

• "What's the difference between the watermark delay and allowed lateness?" — delay defers firing; allowed lateness re-fires an already-fired window for stragglers within grace.
• "What happens to an event later than the grace period?" — side output / dead-letter; never dropped.
• "Why must the sink be idempotent?" — because a window can fire multiple times (initial + late updates); upsert on (window, key).
• "Accumulating vs discarding firing?" — accumulating emits the full corrected aggregate; discarding emits only the delta.
• "How do you tune the grace period?" — from the lateness tail beyond the watermark p99, bounded by how much state you can hold.

Worked example — a window that re-fires on a late event

Detailed explanation. Take the tumbling minute-count from section 3. It fires at count 4 when the watermark passes the window end. Then a late event (still within the grace period) arrives — the window re-fires at count 5. Build the retain-and-refire logic.

• Window. [10:00, 10:01), tumbling, fires at count 4 on the watermark.
• Grace. allowed_lateness = 5m; state retained until 10:06.
• Late event. Event stamped 10:00:45 arrives at watermark 10:02 (within grace) → re-fire at count 5.

Question. Show the window firing on time, then re-firing when a late-but-within-grace event arrives, and specify what the sink does.

Input.

Time	Watermark	Event arriving	Window count
10:01:02	10:01:00	— (fire trigger)	4 → FIRE
10:02:00	10:02:00	ts 10:00:45 (late)	5 → RE-FIRE
10:06:00	10:06:00	— (purge)	state discarded

Code.

WINDOW_START, WINDOW_END = 0, 60      # [10:00, 10:01) in seconds
ALLOWED_LATENESS = 5 * 60             # 5 minutes grace

class TumblingWindowWithLateness:
    def __init__(self):
        self.count = 0
        self.fired = False
        self.purged = False

    def on_event(self, event_ts: int, watermark: int, sink) -> None:
        if self.purged:
            sink.side_output(event_ts)                 # too late: dead-letter
            return
        if WINDOW_START <= event_ts < WINDOW_END:
            self.count += 1
            if self.fired:                             # late-but-within-grace update
                sink.upsert(("win", WINDOW_START), self.count)   # RE-FIRE
        # purge once watermark passes end + grace
        if watermark >= WINDOW_END + ALLOWED_LATENESS:
            self.purged = True

    def on_watermark(self, watermark: int, sink) -> None:
        if not self.fired and watermark >= WINDOW_END:
            self.fired = True
            sink.upsert(("win", WINDOW_START), self.count)       # FIRE
        if watermark >= WINDOW_END + ALLOWED_LATENESS:
            self.purged = True

Step-by-step explanation.

1. The window accumulates on-time events (count reaches 4). When on_watermark sees the watermark reach the window end (60), it fires: sink.upsert(("win", 0), 4). Using upsert from the very first fire means re-fires are just repeat upserts — no special-casing.
2. A late event stamped 10:00:45 arrives while the watermark is at 10:02 — past the window end (fired) but within the 5-minute grace (state not yet purged). on_event increments the count to 5 and, because the window has already fired, immediately re-fires with sink.upsert(("win", 0), 5).
3. The sink receives two writes for the same key ("win", 0): first value 4, then value 5. Because it upserts, the final stored value is 5 — the corrected count. An append-only sink would have stored both and double-counted; the idempotent upsert is what makes re-firing safe.
4. When the watermark finally passes window_end + allowed_lateness (60 + 300 = 360 = 10:06), the window purges its state. Any event arriving after that is routed to sink.side_output(...) — it can no longer correct the window because the state is gone.
5. This is the full lateness contract in one object: on-time events fire once, within-grace late events re-fire corrections via upsert, past-grace events go to the side output. Nothing is dropped; the only difference is where a late event is reflected — in the window (within grace) or in the reconciliation channel (past grace).

Output.

Fire #	Trigger	Value written to sink	Stored (upsert)
1	watermark ≥ 10:01	count 4	4
2	late event within grace	count 5	5 (replaces 4)
—	event past 10:06	side output	not in window

Rule of thumb. Allowed lateness = retain window state past its fire so late-but-within-grace events re-fire corrected results; the sink must upsert on (window, key) so re-fires replace rather than append. Past the grace, route to a side output — the window can no longer help.

Worked example — routing too-late events to a side output

Detailed explanation. The side output is the dead-letter channel for events past the grace period. It must be observable — you alert on its volume and reconcile from it in batch. Build the routing and a Kafka dead-letter topic, then show the reconciliation query.

• Trigger. Event arrives after window_end + allowed_lateness.
• Destination. A dead-letter Kafka topic (or table) capturing the raw event plus why it was late.
• Reconciliation. A nightly batch job folds dead-letter events into corrected aggregates.

Question. Route past-grace events to a dead-letter sink with enough metadata to reconcile, and write the reconciliation query.

Input.

Field	Value
Dead-letter topic	late-events-dlq
Captured metadata	event, event_ts, window_start, watermark_at_drop, lateness_s
Reconcile cadence	nightly
Alert	dlq volume > baseline

Code.

import json

def route_late_event(event: dict, window_start: int, window_end: int,
                     allowed_lateness: int, watermark: int, dlq_producer) -> None:
    """Send a past-grace event to the dead-letter side output with context."""
    lateness_s = watermark - window_end
    record = {
        "event":              event,
        "event_ts":           event["ts"],
        "window_start":       window_start,
        "watermark_at_drop":  watermark,
        "lateness_seconds":   lateness_s,
        "reason":             "past_allowed_lateness",
    }
    # Never drop: the DLQ is the visible, reconcilable record of lateness.
    dlq_producer.send("late-events-dlq", json.dumps(record).encode())

-- Nightly reconciliation — fold dead-letter events into corrected aggregates.
-- Idempotent: recomputes each affected window from scratch and upserts.
WITH dlq AS (
    SELECT (payload->>'event_ts')::bigint      AS event_ts,
           (payload->'event'->>'key')          AS key,
           (payload->>'window_start')::bigint  AS window_start
    FROM   late_events_dlq
    WHERE  ingested_at::date = current_date - 1
),
recount AS (
    SELECT window_start, key, count(*) AS late_extra
    FROM   dlq
    GROUP  BY window_start, key
)
UPDATE window_results w
SET    cnt = w.cnt + r.late_extra,
       corrected_at = now()
FROM   recount r
WHERE  w.window_start = r.window_start
  AND  w.key = r.key;

Step-by-step explanation.

1. route_late_event fires only for events past the grace period. It captures not just the raw event but the context — which window it belonged to, the watermark at the moment it was dropped, and the exact lateness in seconds. That context is what makes the dead-letter reconcilable rather than just a graveyard.
2. The DLQ is a first-class Kafka topic (or table), durable and queryable. The key design decision is that a too-late event is never passed to /dev/null — it always lands somewhere you can see it, count it, and act on it.
3. Alerting on DLQ volume is the operational payoff: a sudden spike in lateness_seconds or DLQ throughput is an early signal that a source is misbehaving (a region's clocks drifted, a partition backed up) — often before the main metric visibly degrades.
4. The nightly reconciliation recomputes each affected (window_start, key) by folding the dead-letter counts back in and upserting the corrected totals. Because it recomputes and upserts (rather than blindly incrementing), the job is idempotent — re-running it does not double-correct.
5. The end state is a two-tier correctness model: the streaming layer is correct within the grace period at low latency, and the batch reconciliation layer catches the long tail with a day's delay. This is the lambda-style "fast approximate + slow exact" split applied specifically to lateness.

Output.

Stage	Latency	Correctness
Streaming (within grace)	seconds	exact for on-time + within-grace
Side output capture	immediate	records every too-late event
Nightly reconciliation	~1 day	folds in the long tail
Silently dropped	—	zero (the whole point)

Rule of thumb. Past-grace events go to an observable side output with full context (window, watermark, lateness), never to /dev/null. Alert on its volume and reconcile from it nightly — a visible 0.1% you correct beats an invisible 0.1% you never learn about.

Worked example — the idempotent late-update sink

Detailed explanation. Because a window can fire multiple times (initial fire plus late-within-grace re-fires), the sink write must be idempotent on (window, key). A naive append double-counts; the correct pattern is an upsert. Build the sink and show what happens under a re-fire with both a correct and an incorrect sink.

• Key. (window_start, key) — the identity of the result being written.
• Correct. INSERT ... ON CONFLICT (window_start, key) DO UPDATE — upsert, last-write-wins.
• Incorrect. INSERT — appends a second row on re-fire, double-counting downstream.

Question. Implement an idempotent sink for windowed results and contrast it with the naive append under a late re-fire.

Input.

Sink write	window_start	key	value	Correct sink result	Naive sink result
initial fire	0	user_7	4	row=4	row=4
late re-fire	0	user_7	5	row=5 (updated)	rows 4 AND 5 (double)

Code.

-- Correct: idempotent upsert keyed on (window_start, key)
CREATE TABLE window_results (
    window_start BIGINT NOT NULL,
    key          TEXT   NOT NULL,
    cnt          BIGINT NOT NULL,
    updated_at   TIMESTAMPTZ NOT NULL DEFAULT now(),
    PRIMARY KEY (window_start, key)
);

-- Every fire (initial or re-fire) issues the SAME statement:
INSERT INTO window_results (window_start, key, cnt)
VALUES (0, 'user_7', 5)
ON CONFLICT (window_start, key)
DO UPDATE SET cnt = EXCLUDED.cnt, updated_at = now();
-- initial fire stored 4; the re-fire replaces it with 5. Final: one row, cnt=5.

# The sink object the window fires into — one method, always idempotent
class IdempotentUpsertSink:
    def __init__(self, conn):
        self.conn = conn

    def upsert(self, window_key: tuple[int, str], value: int) -> None:
        window_start, key = window_key
        with self.conn.cursor() as cur:
            cur.execute("""
                INSERT INTO window_results (window_start, key, cnt)
                VALUES (%s, %s, %s)
                ON CONFLICT (window_start, key)
                DO UPDATE SET cnt = EXCLUDED.cnt, updated_at = now()
            """, (window_start, key, value))
        self.conn.commit()

    def side_output(self, event) -> None:
        # past-grace events go here (see previous example)
        ...

Step-by-step explanation.

1. The window_results table's primary key is (window_start, key) — the identity of a windowed result. This is the linchpin: the PK is what lets the database recognise a re-fire as an update to an existing result rather than a new one.
2. The initial fire inserts (0, 'user_7', 4). The late re-fire runs the identical statement with value 5. The ON CONFLICT ... DO UPDATE clause detects the existing row and overwrites cnt to 5. The final state is one row with the corrected count.
3. Contrast the naive INSERT: the initial fire stores 4, and the re-fire stores a second row with 5. A downstream SUM(cnt) now reads 9 instead of 5 — the late correction turned into a double-count. This is the single most common late-data bug in production.
4. Because every fire — first or last — issues the same upsert, the window logic does not need to know whether it is firing for the first time or re-firing. Idempotency is pushed entirely into the sink, keeping the window code simple and correct by construction.
5. This closes the loop opened by allowed lateness: re-firing is only safe because the sink is idempotent. Allowed lateness (retain + re-fire) and the idempotent sink (upsert) are two halves of one contract — you cannot correctly have one without the other.

Output.

After	Correct sink (upsert)	Naive sink (insert)
initial fire	1 row: cnt=4	1 row: cnt=4
late re-fire	1 row: cnt=5	2 rows: 4 and 5
downstream SUM	5 (correct)	9 (double-counted)

Rule of thumb. Any sink that receives windowed results must upsert on (window, key) because allowed lateness guarantees windows re-fire. An append-only sink silently double-counts every late correction — make idempotency the sink's invariant, not the window's problem.

Data engineering interview question on allowed lateness

A senior interviewer might ask: "You're building a real-time revenue aggregation by minute from a payments stream where about 2% of events arrive late — most within a couple of minutes, a few up to an hour. Finance needs the dashboard to self-correct as late payments land, and absolutely nothing can be dropped. Design the lateness handling — the watermark delay, the allowed lateness, the sink contract, and the side-output reconciliation — and explain the completeness guarantee you can promise."

Solution Using watermark delay + allowed lateness + idempotent upsert + dead-letter reconciliation

# Minute revenue aggregation with the full lateness contract.
WINDOW_S         = 60
WATERMARK_DELAY  = 10          # covers p99 out-of-orderness
ALLOWED_LATENESS = 5 * 60      # re-fire corrections up to 5 min late

def window_start(ts: int) -> int:
    return (ts // WINDOW_S) * WINDOW_S

class RevenueWindows:
    def __init__(self, sink):
        self.revenue: dict[int, float] = {}     # window_start -> revenue cents
        self.fired: set[int] = set()
        self.sink = sink

    def on_event(self, ts: int, amount_cents: float, watermark: int) -> None:
        w = window_start(ts)
        if watermark > w + WINDOW_S + ALLOWED_LATENESS:
            self.sink.side_output({"ts": ts, "amount": amount_cents, "window": w})
            return                              # past grace -> dead-letter
        self.revenue[w] = self.revenue.get(w, 0) + amount_cents
        if w in self.fired:                     # late-within-grace -> re-fire
            self.sink.upsert(("rev", w), self.revenue[w])

    def on_watermark(self, watermark: int) -> None:
        for w, rev in list(self.revenue.items()):
            if w not in self.fired and watermark >= w + WINDOW_S:
                self.fired.add(w)
                self.sink.upsert(("rev", w), rev)          # first fire
            if watermark > w + WINDOW_S + ALLOWED_LATENESS:
                del self.revenue[w]                        # purge -> bounded state

-- Idempotent revenue sink + nightly dead-letter reconciliation
CREATE TABLE revenue_by_minute (
    window_start BIGINT PRIMARY KEY,
    revenue_cents BIGINT NOT NULL,
    corrected_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- The streaming upsert (every fire, first or re-fire):
-- INSERT ... ON CONFLICT (window_start) DO UPDATE SET revenue_cents = EXCLUDED.revenue_cents;

-- Nightly: fold past-grace dead-letter payments back in, idempotently.
UPDATE revenue_by_minute r
SET    revenue_cents = base.on_time + COALESCE(dlq.late_sum, 0),
       corrected_at  = now()
FROM  (SELECT window_start, sum(amount) AS on_time FROM payments_agg GROUP BY 1) base
LEFT  JOIN (SELECT (payload->>'window')::bigint AS window_start,
                   sum((payload->>'amount')::bigint) AS late_sum
            FROM late_events_dlq GROUP BY 1) dlq
       ON dlq.window_start = base.window_start
WHERE r.window_start = base.window_start;

Step-by-step trace.

Layer	Mechanism	Guarantee
Watermark delay 10s	defer fire for p99 out-of-orderness	first fire is nearly complete
First fire	upsert on watermark ≥ window end	dashboard shows result in ~10s
Within-grace late (≤5m)	retain state, re-fire via upsert	dashboard self-corrects
Idempotent sink	ON CONFLICT DO UPDATE	re-fires replace, never double-count
Past-grace (>5m)	side output to DLQ	captured, never dropped
Nightly reconcile	fold DLQ, recompute, upsert	eventual exactness

After deployment, each minute's revenue appears on the dashboard about ten seconds after the minute closes, then self-corrects upward as late payments land within the five-minute grace — each correction an idempotent upsert that replaces the prior value. Payments later than five minutes flow to the late-events-dlq, where a nightly job folds them into the final figure. The completeness guarantee is precise: exact within five minutes of latency in real time, and eventually exact (by the next day) for the long tail, with zero silent drops.

Output:

Property	Guarantee
Real-time latency	~10 s after minute close
Self-correction window	5 min (allowed lateness)
Silent drops	zero (side output)
Eventual exactness	next-day reconciliation
Double-count risk	zero (idempotent upsert)

Why this works — concept by concept:

• Watermark delay for the common case — a 10-second delay covers p99 out-of-orderness so the first fire is nearly complete without paying max latency; the rarer tail is delegated to lateness rather than a bigger delay.
• Allowed lateness re-fires — retaining window state for five minutes lets late payments update and re-emit the corrected revenue, giving finance a dashboard that self-heals instead of one that's quietly wrong.
• Idempotent upsert sink — because every fire (first or re-fire) is an ON CONFLICT DO UPDATE, corrections replace rather than accumulate, eliminating the double-count that append-only sinks suffer on re-fire.
• Side output + nightly reconcile — past-grace payments are captured in a dead-letter channel and folded back in by an idempotent batch job, converting "we can't wait an hour in-stream" into "we're exact by tomorrow" instead of "we lose it."
• Cost — O(open windows) revenue state held for window + delay + lateness (~5 min), plus a dead-letter topic and one nightly batch pass. The trade is a few minutes of retained state and a one-day tail correction in exchange for a hard no-silent-drop guarantee — exactly the contract finance requires.

Streaming
Topic — event-processing
Allowed-lateness and dead-letter problems

Practice →

Streaming Topic — streaming Idempotent-sink and correction problems

Practice →

---

5. Reordering & correctness patterns

Out-of-order is a sequence problem, not a lateness problem — you solve it with bounded sort buffers, gap detection, and idempotent keyed dedupe, all bounded by the watermark

The mental model in one line: reordering is the family of patterns that restore sequence correctness to a stream where events arrive in the wrong order — a bounded sort buffer that emits events in event-time order once the watermark guarantees no earlier event can still arrive, sequence-number gap detection that flags missing records, and idempotent keyed dedupe that makes replay and at-least-once delivery safe — and the unifying discipline is that all of this state is bounded by the watermark, so an unbounded stream never consumes unbounded memory. Lateness (section 4) is about when relative to the watermark; reordering is about sequence relative to other events. The two interact — late events are always out of order — but the patterns are distinct, and senior interviews probe whether you can tell them apart.

The three reordering patterns — and what each guarantees.

• Bounded sort buffer. Buffer incoming events and emit them in event-time order, releasing an event only once the watermark proves no earlier event can still arrive. Guarantees ordered output within the watermark bound. The cost is latency equal to the watermark delay and state proportional to events-in-flight.
• Sequence-gap detection. When events carry a monotonic per-key sequence number, detect missing numbers (a gap) to distinguish "out of order but complete" from "genuinely lost." Guarantees completeness detection — you learn when a record is missing rather than silently proceeding.
• Idempotent keyed dedupe. Deduplicate by a stable event key (or (key, sequence)), so at-least-once delivery and replays do not double-process. Guarantees exactly-once effect even over at-least-once transport. State is bounded by retaining dedupe keys only within the watermark horizon.

Bounded sort buffer — ordering within a bound.

• The buffer. A min-heap (or sorted structure) keyed by event time, holding events not yet safe to emit.
• The release rule. Emit every buffered event with event_time <= watermark — the watermark guarantees nothing earlier will arrive, so those events are now in final order.
• The bound. Buffer size is proportional to how far events are out of order (the watermark delay). A larger delay = more ordering tolerance = more buffered state and more latency.

Sequence-gap detection — knowing when something is missing.

• The precondition. The source stamps a per-key monotonic sequence number (Kafka offsets per partition, a DB LSN, an app-level counter). Without a monotone key, gaps are undetectable.
• The detector. Track the highest contiguous sequence seen per key; a jump (e.g. ...5, 6, 8... with 7 missing) is a gap. A gap may be transient (7 is merely out of order and arrives soon) or permanent (7 is lost).
• The resolution. Wait a bounded time (the watermark / a timeout) for the gap to fill; if it does not, alert / reconcile. This is how you turn "silent data loss" into "detected, alertable loss."

Idempotent keyed dedupe — exactly-once over at-least-once.

• The reality. Most transports are at-least-once: Kafka redelivers on rebalance, a producer retries on timeout, a replay re-emits. Duplicates are normal.
• The dedupe. Maintain a set of seen (key, sequence) (or a unique event id); skip anything already seen. Wrap the dedupe check and the side effect in one transaction so "seen" and "done" commit together.
• The bound. You cannot remember every key forever. Retain dedupe state only within the watermark horizon (plus allowed lateness) — beyond that, no duplicate can still legitimately arrive, so the key can be evicted. This is what makes exactly-once bounded.

Common interview probes on reordering.

• "Out-of-order vs late — what's the difference?" — out-of-order is about arrival sequence; late is about the watermark. All late events are out of order; most out-of-order events aren't late.
• "How do you emit a stream in order?" — bounded sort buffer released by the watermark.
• "How do you detect a lost event?" — per-key monotonic sequence + gap detection + a bounded wait.
• "How do you get exactly-once over Kafka?" — idempotent keyed dedupe with watermark-bounded state, side effect and dedupe committed atomically.
• "What bounds the dedupe state?" — the watermark horizon plus allowed lateness; evict keys older than that.

Worked example — a bounded reorder buffer released by the watermark

Detailed explanation. The canonical reordering primitive: a min-heap buffer that holds out-of-order events and releases them in event-time order once the watermark guarantees no earlier event can arrive. Build it and trace an out-of-order stream becoming ordered output.

• Buffer. A min-heap keyed by event time.
• On event. Push into the heap.
• On watermark. Pop and emit every event with event_time <= watermark, in order.

Question. Implement a watermark-released reorder buffer and trace it turning an out-of-order stream into ordered output.

Input.

Arrival #	event_ts	watermark after	emitted (ts ≤ wm)
1	5	2	—
2	4	2	—
3	9	6	4, 5
4	7	6	—

Code.

import heapq

class ReorderBuffer:
    def __init__(self):
        self.heap: list[int] = []     # min-heap of event times

    def on_event(self, event_ts: int) -> None:
        heapq.heappush(self.heap, event_ts)

    def on_watermark(self, watermark: int) -> list[int]:
        """Emit, in order, all buffered events the watermark has made safe."""
        emitted = []
        while self.heap and self.heap[0] <= watermark:
            emitted.append(heapq.heappop(self.heap))
        return emitted


buf = ReorderBuffer()
# (event_ts, watermark_after) with delay = 3, watermark = max_seen - 3
stream = [(5, 2), (4, 2), (9, 6), (7, 6)]
for ts, wm in stream:
    buf.on_event(ts)
    out = buf.on_watermark(wm)
    print(f"in={ts} wm={wm} emit={out} buffered={sorted(buf.heap)}")
# in=5 wm=2 emit=[]     buffered=[5]
# in=4 wm=2 emit=[]     buffered=[4, 5]
# in=9 wm=6 emit=[4, 5] buffered=[7, 9]
# in=7 wm=6 emit=[]     buffered=[7, 9]

Step-by-step explanation.

1. Each arriving event is pushed onto a min-heap ordered by event time. The heap holds events that have arrived but are not yet safe to emit, because an earlier event might still be in flight.
2. After arrivals 1 and 2 (event times 5 and 4), the watermark is only at 2, so nothing is safe to release — an event with time 3 could still arrive. The buffer holds [4, 5], out of arrival order but heap-ordered.
3. When arrival 3 (event time 9) pushes the watermark to 6, the buffer releases every event with event_time <= 6: that is 4 then 5, emitted in event-time order even though 5 arrived before 4. The watermark is the correctness guarantee — at wm 6, nothing ≤ 6 can still come.
4. Arrival 4 (event time 7) does not advance the watermark (max_seen is still 9), so 7 and 9 stay buffered until the watermark reaches them. The buffer is self-bounding: its size tracks the out-of-orderness the watermark delay tolerates.
5. The output is a totally ordered event-time stream, at the cost of latency equal to the watermark delay. This is the reordering-versus-latency trade, exactly parallel to the completeness-versus-latency trade of the watermark itself — ordering, like completeness, is bought with delay.

Output.

Watermark	Emitted (in order)	Still buffered
2	—	4, 5
6	4, 5	7, 9
9 (later)	7, 9	—

Rule of thumb. A bounded sort buffer released by the watermark converts an out-of-order stream into an ordered one, with latency equal to the watermark delay and state bounded by the out-of-orderness. Order, like completeness, costs exactly the watermark delay — no more, no less.

Worked example — sequence-gap detection

Detailed explanation. When events carry a per-key monotonic sequence number, you can distinguish "out of order but complete" from "a record is genuinely missing." The detector tracks the highest contiguous sequence and flags gaps, waiting a bounded time for out-of-order fills before alerting. Build it.

• Precondition. Each event has (key, seq) with seq monotonic per key.
• Detector. Track contiguous_high[key]; buffer out-of-order seqs; advance when gaps fill.
• Gap. A seq beyond contiguous_high + 1 with the intermediate seqs missing → potential gap.

Question. Detect a missing sequence number in an out-of-order stream and distinguish a transient gap (fills soon) from a permanent one (lost).

Input.

Arrival #	seq	contiguous_high after	gap?
1	5	5	no
2	6	6	no
3	8	6	7 pending
4	7	8	filled

Code.

class SequenceGapDetector:
    def __init__(self):
        self.contiguous_high: int | None = None
        self.pending: set[int] = set()      # seqs seen ahead of the contiguous high

    def on_event(self, seq: int) -> None:
        if self.contiguous_high is None:
            self.contiguous_high = seq
            return
        if seq <= self.contiguous_high:
            return                          # duplicate / already covered
        self.pending.add(seq)
        # advance the contiguous high as long as the next seq is present
        while (self.contiguous_high + 1) in self.pending:
            self.contiguous_high += 1
            self.pending.discard(self.contiguous_high)

    def gaps(self) -> list[int]:
        """Sequence numbers still missing below the highest seen."""
        if self.contiguous_high is None or not self.pending:
            return []
        highest = max(self.pending)
        return [s for s in range(self.contiguous_high + 1, highest) if s not in self.pending]


det = SequenceGapDetector()
for seq in [5, 6, 8, 7]:
    det.on_event(seq)
    print(f"seq={seq} contiguous_high={det.contiguous_high} gaps={det.gaps()}")
# seq=5 contiguous_high=5 gaps=[]
# seq=6 contiguous_high=6 gaps=[]
# seq=8 contiguous_high=6 gaps=[7]   <- 7 missing, 8 pending: transient gap
# seq=7 contiguous_high=8 gaps=[]    <- 7 arrived out of order; gap filled

Step-by-step explanation.

1. The detector tracks contiguous_high — the highest sequence such that every seq up to it has been seen. Anything beyond it that arrives early is held in pending, waiting for the intervening seqs.
2. Seqs 5 and 6 arrive in order; contiguous_high advances to 6 with no gap. When seq 8 arrives, it is beyond contiguous_high + 1 = 7, so 8 goes into pending and the detector reports a gap at 7 — 7 is missing so far.
3. This gap is initially ambiguous: 7 might be merely out of order (transient) or genuinely lost (permanent). The detector does not decide yet; it reports the gap and waits.
4. Seq 7 then arrives out of order. The while loop advances contiguous_high from 6 to 7 (7 now present), then to 8 (already in pending). The gap is filled; gaps() returns empty. The out-of-order 7 was transient.
5. The resolution rule in production: pair the detector with a bounded wait tied to the watermark. If the gap has not filled by the time the watermark passes the gap's event time (i.e. no earlier event can still arrive), the gap is permanent — alert and reconcile. This is how sequence numbers plus watermarks turn silent loss into detected, actionable loss.

Output.

seq seen	contiguous_high	gaps	verdict
5, 6	6	[]	complete
8	6	[7]	7 pending (transient?)
7	8	[]	gap filled

Rule of thumb. Per-key monotonic sequence numbers plus a gap detector turn undetectable data loss into an alertable event. Give each gap a bounded wait (the watermark); if it does not fill by then, it is permanent — reconcile it rather than proceeding as if the stream were complete.

Worked example — idempotent keyed dedupe for exactly-once

Detailed explanation. At-least-once transports (Kafka, retrying producers, replays) deliver duplicates. Idempotent keyed dedupe makes the effect exactly-once by skipping already-seen keys, with state bounded by the watermark so it does not grow forever. Build the dedupe with atomic side-effect commit and watermark-bounded eviction.

• Dedupe key. A stable unique id per event ((key, seq) or an event UUID).
• Atomicity. Record "seen" and perform the side effect in one transaction.
• Bound. Evict dedupe keys older than the watermark horizon (+ allowed lateness) — beyond that, no duplicate can legitimately arrive.

Question. Implement idempotent keyed dedupe that survives replay and bounds its state by the watermark.

Input.

Arrival #	event_id	event_ts	seen before?	action
1	e-100	10	no	process
2	e-101	12	no	process
3	e-100	10	yes (replay)	skip
4	e-102	40	no	process; evict < wm

Code.

class WatermarkBoundedDedupe:
    def __init__(self, retention_s: int):
        self.seen: dict[str, int] = {}     # event_id -> event_ts
        self.retention = retention_s        # keep keys within this of the watermark

    def process(self, event_id: str, event_ts: int, watermark: int, do_effect) -> bool:
        """Return True if processed, False if skipped as a duplicate."""
        if event_id in self.seen:
            return False                    # duplicate: exactly-once effect preserved
        # atomic in a real sink: record 'seen' AND run the side effect together
        do_effect(event_id, event_ts)
        self.seen[event_id] = event_ts
        self._evict(watermark)
        return True

    def _evict(self, watermark: int) -> None:
        cutoff = watermark - self.retention
        # no duplicate older than the cutoff can still legitimately arrive
        for eid in [k for k, ts in self.seen.items() if ts < cutoff]:
            del self.seen[eid]


dd = WatermarkBoundedDedupe(retention_s=20)
log = []
stream = [("e-100", 10, 7), ("e-101", 12, 9), ("e-100", 10, 9), ("e-102", 40, 37)]
for eid, ts, wm in stream:
    processed = dd.process(eid, ts, wm, lambda i, t: log.append(i))
    print(f"{eid} ts={ts} wm={wm} processed={processed} keys={sorted(dd.seen)}")
# e-100 ts=10 wm=7  processed=True  keys=['e-100']
# e-101 ts=12 wm=9  processed=True  keys=['e-100', 'e-101']
# e-100 ts=10 wm=9  processed=False keys=['e-100', 'e-101']   <- replay skipped
# e-102 ts=40 wm=37 processed=True  keys=['e-102']            <- old keys evicted (wm-20=17)

Step-by-step explanation.

1. The dedupe keeps a map of event_id -> event_ts for every event it has processed. On each arrival it checks membership: a known id is a duplicate and is skipped, guaranteeing the side effect runs at most once per id.
2. Arrivals 1 and 2 (e-100, e-101) are new, so they process and are recorded. Arrival 3 is e-100 again — a replay or a Kafka redelivery. It is found in seen, so it is skipped; the side effect does not run twice. This is the exactly-once effect over at-least-once delivery.
3. In a real sink, the "record seen" and "do the side effect" must be one transaction — otherwise a crash between them either double-processes (effect ran, not recorded) or loses the event (recorded, effect didn't run). Atomicity is what makes the dedupe correct under failure, not just under duplicates.
4. The eviction step is what makes exactly-once bounded. When e-102 arrives at event time 40 with watermark 37, the cutoff is 37 - 20 = 17. Keys with event time below 17 (e-100, e-101) are evicted — the watermark guarantees no duplicate of those can still arrive, so remembering them is wasted state.
5. The retention must be at least the watermark horizon plus allowed lateness: any event that could still legitimately re-arrive (late, within grace) must still find its key. Set retention below that and a late duplicate slips through; set it far above and you hold needless state. It ties directly to the lateness configuration of section 4.

Output.

event_id	processed?	reason	keys retained
e-100 (1st)	yes	new	e-100
e-101	yes	new	e-100, e-101
e-100 (2nd)	no	duplicate/replay	e-100, e-101
e-102	yes	new; evict < 17	e-102

Rule of thumb. Exactly-once = idempotent keyed dedupe with the "seen" record and the side effect committed atomically, and dedupe state evicted at the watermark horizon plus allowed lateness. Remember keys exactly as long as a duplicate could still legitimately arrive — no longer.

Data engineering interview question on reordering and exactly-once

A senior interviewer might ask: "You consume an out-of-order, partitioned Kafka stream with at-least-once delivery. Downstream needs events processed in event-time order, exactly once, with any genuinely-lost event detected rather than silently skipped. Design the full pipeline — the reorder buffer, the gap detection, the dedupe, and how the watermark bounds every piece of state — and explain the ordering, completeness, and exactly-once guarantees you can make."

Solution Using a watermark-bounded reorder buffer + gap detection + idempotent dedupe

# Composed reordering pipeline: dedupe -> gap-detect -> reorder -> emit in order.
import heapq

class OrderedExactlyOncePipeline:
    def __init__(self, dedupe_retention_s: int):
        self.seen: dict[str, int] = {}          # event_id -> ts (dedupe)
        self.dedupe_retention = dedupe_retention_s
        self.contiguous_high: dict[str, int] = {}   # key -> contiguous seq high
        self.pending_seq: dict[str, set[int]] = {}  # key -> seqs seen ahead
        self.buffer: list[tuple[int, str, dict]] = []   # min-heap by event_ts

    def on_event(self, ev: dict, watermark: int) -> None:
        eid, key, seq, ts = ev["id"], ev["key"], ev["seq"], ev["ts"]
        # 1. dedupe (exactly-once effect over at-least-once transport)
        if eid in self.seen:
            return
        self.seen[eid] = ts
        # 2. gap detection (completeness) per key
        self._track_seq(key, seq)
        # 3. reorder buffer (ordering) keyed by event time
        heapq.heappush(self.buffer, (ts, eid, ev))

    def _track_seq(self, key: str, seq: int) -> None:
        hi = self.contiguous_high.get(key)
        if hi is None:
            self.contiguous_high[key] = seq
            return
        pend = self.pending_seq.setdefault(key, set())
        if seq > hi:
            pend.add(seq)
            while (self.contiguous_high[key] + 1) in pend:
                self.contiguous_high[key] += 1
                pend.discard(self.contiguous_high[key])

    def on_watermark(self, watermark: int, emit) -> None:
        # release ordered events the watermark has made safe
        while self.buffer and self.buffer[0][0] <= watermark:
            _, _, ev = heapq.heappop(self.buffer)
            emit(ev)
        # bound dedupe state by the watermark horizon
        cutoff = watermark - self.dedupe_retention
        for eid in [k for k, t in self.seen.items() if t < cutoff]:
            del self.seen[eid]

    def permanent_gaps(self, key: str, watermark: int) -> list[int]:
        pend = self.pending_seq.get(key, set())
        if not pend:
            return []
        hi = self.contiguous_high[key]
        return list(range(hi + 1, max(pend)))   # unfilled below the highest seen

Step-by-step trace.

Stage	Mechanism	Guarantee produced
1. Dedupe	skip seen event_id	exactly-once effect
2. Gap detect	per-key contiguous seq + pending	completeness detection
3. Reorder buffer	min-heap by event_ts	event-time ordered output
4. Watermark release	emit ts ≤ watermark	correct order, bounded latency
5. Dedupe eviction	drop keys < wm − retention	bounded state
6. Permanent gap	unfilled seq below highest at watermark	detected loss, alertable

After deployment, each Kafka event is first deduplicated by id (so redeliveries and replays are no-ops), then its per-key sequence is tracked so a missing offset is flagged, then it is buffered and released in strict event-time order once the watermark proves no earlier event can arrive. Every piece of state — dedupe keys, reorder buffer, pending sequences — is bounded by the watermark horizon, so the pipeline runs in constant memory relative to the out-of-orderness bound regardless of total stream volume. Genuinely-lost events surface as permanent gaps once the watermark passes them, turning silent loss into an alert.

Output:

Guarantee	How	Bound
Ordering	reorder buffer released by watermark	latency = watermark delay
Exactly-once	idempotent keyed dedupe	state = keys within horizon
Completeness	per-key sequence gap detection	gap resolved at watermark
Bounded state	watermark-driven eviction	O(events in flight)

Why this works — concept by concept:

• Idempotent keyed dedupe — skipping already-seen event ids gives an exactly-once effect over at-least-once Kafka delivery, so redeliveries and replays never double-process.
• Per-key sequence-gap detection — tracking the contiguous sequence high per key converts an out-of-order arrival into either a transient (soon-filled) gap or, at the watermark, a permanent detected loss — no silent skips.
• Watermark-released reorder buffer — a min-heap keyed by event time that emits only events at or below the watermark produces a totally ordered output stream, with latency exactly the watermark delay.
• Watermark-bounded eviction — evicting dedupe keys and releasing buffered events at the watermark horizon is what keeps every structure bounded, so the pipeline uses constant memory relative to the out-of-orderness, not to total volume.
• Cost — O(events in flight within the watermark delay) for the buffer, O(distinct keys within the horizon) for dedupe, and O(pending seqs per key) for gap detection — all bounded by the watermark. The trade is watermark-delay latency and horizon-sized state in exchange for simultaneous ordering, exactly-once, and completeness guarantees over an out-of-order at-least-once source.

Streaming
Topic — event-processing
Reordering and exactly-once problems

Practice →

Streaming
Topic — time-series
Sequence-gap and ordered-delivery problems

Practice →

---

Cheat sheet — late & out-of-order data recipes

• Event-time vs processing-time decision rule. If the output will ever be audited against a system of record or recomputed via replay, compute it in event time — only event time is reproducible. Use processing time solely for latency-only live gauges no one audits; use ingestion time as a documented fallback when the source carries no trustworthy embedded timestamp (it misattributes buffered data). Fix the time domain before touching windowing.
• Late vs out-of-order. Late is watermark-relative (arrived after its slot was declared complete); out-of-order is arrival-relative (arrived after a later-timestamped event). All late events are out of order; most out-of-order events are not late. Handle out-of-orderness with the watermark delay; handle lateness with allowed lateness + side outputs.
• Watermark generator template. watermark = max_event_time_seen − delay (bounded-out-of-orderness), monotonic (never regresses). Set delay from the measured p99 skew (received_at − event_time), not the max. The p99-to-max gap is the population you handle with allowed lateness and side outputs, not with a bigger delay.
• Watermark propagation + idleness. An operator's output watermark is the minimum of its input watermarks and never regresses; one slow input limits the whole graph. Always configure an idleness timeout on multi-partition sources so a silent partition is excluded from the min after N seconds — otherwise one quiet partition freezes every window with no error. Idle sources that wake up may emit late events; handle, don't drop.
• Window-type selection. Tumbling (fixed, non-overlapping, one event → one window) for disjoint periodic aggregates ("per minute/hour/day"). Sliding (fixed size, smaller slide, one event → size/slide windows, overlap = size − slide) for rolling/smoothed metrics ("last 5 min every 1 min"). Session (variable, gap-defined, mergeable) for activity bursts ("per visit/trip"). All fire on the watermark and purge after allowed lateness.
• Window lifecycle. assign (by event time) → accumulate (per open window per key) → trigger (when watermark passes window end) → fire (emit) → purge (after allowed lateness). State is O(open windows × keys); purge is what bounds memory on an unbounded stream.
• Allowed lateness template. Retain window state for allowed_lateness past its fire; a late-but-within-grace event updates the aggregate and re-fires a corrected result. Sink must upsert on (window, key) (accumulating mode) so re-fires replace rather than double-count. Grace picks up where the watermark delay ends; set it from the lateness tail you can afford to hold state for.
• Side output / dead-letter rule. Events past window_end + allowed_lateness go to an observable side output with full context (event, window, watermark-at-drop, lateness seconds) — never to /dev/null. Alert on its volume (spikes signal a source problem) and fold it into a nightly idempotent reconciliation. Guarantee: exact within grace, eventually exact via reconciliation, dropped never.
• Idempotent sink invariant. Because allowed lateness makes windows fire multiple times, every windowed-result sink must be idempotent on (window, key) — INSERT ... ON CONFLICT DO UPDATE. Append-only sinks silently double-count every late correction. Push idempotency into the sink, not the window logic.
• Reorder buffer template. Min-heap keyed by event time; on watermark, emit every buffered event with event_time ≤ watermark in order. Produces a totally ordered stream at latency = watermark delay and state proportional to the out-of-orderness. Ordering, like completeness, costs exactly the watermark delay.
• Sequence-gap detection. Requires a per-key monotonic sequence (Kafka offset, DB LSN, app counter). Track the contiguous high; a jump is a gap. Give each gap a bounded wait tied to the watermark — if unfilled when the watermark passes, it is a permanent detected loss to alert/reconcile, not a silent skip.
• Exactly-once = dedupe + bounded state. Idempotent keyed dedupe ((key, seq) or event id) makes at-least-once transport exactly-once in effect; commit the "seen" record and the side effect atomically. Evict dedupe keys older than the watermark horizon + allowed lateness — remember a key exactly as long as a duplicate could still legitimately arrive.
• Two-tier completeness. Fast streaming layer = exact within the grace period at low latency; slow batch layer = folds the dead-letter side output into an eventually-exact result by the next day. This lambda-style split, applied to lateness, is how you promise "exact soon, eventually exact for the tail, never lost."

Frequently asked questions

What is late and out-of-order data in one sentence?

Late and out-of-order data is the normal condition of any real event stream in which records do not arrive in the order their event timestamps imply — an event stamped earlier can show up after an event stamped later (out of order), and an event can show up after the pipeline has already declared its event-time slot complete (late). Out-of-order is a statement about arrival sequence; late is a statement about the watermark. Every late event is out of order, but most out-of-order events are not late, because a correctly-set watermark delay is designed to tolerate the expected disorder. Handling both correctly — with event-time semantics, watermarks, windowing, allowed lateness, and reordering — is the load-bearing skill of production stream processing.

Event time vs processing time — when do I pick each?

Pick event time whenever the output will be audited against a system of record or recomputed via replay — billing, revenue, funnel analytics, SLA compliance — because event time is the only clock under which a replay of the same input yields the identical output. Event time buckets each record by the immutable timestamp of when the thing actually happened, so a beacon that buffered offline for an hour is still counted in the correct hour. Pick processing time only for latency-dominated live views that nobody audits — a "requests per second right now" gauge — where the lowest possible latency matters more than reproducibility. Processing time is non-deterministic under replay and systematically misattributes late data to the wrong window, so the moment someone asks "what was it yesterday at 3pm," it has already failed. Ingestion time is a documented fallback when the source has no trustworthy embedded timestamp.

What is a watermark and how do I set the delay?

A watermark is a monotonically advancing timestamp that flows through the stream and asserts "I believe I have now seen every event with an event time at or before W" — it is the completeness signal that triggers event-time windows to fire. The standard generator is bounded-out-of-orderness: watermark = max_event_time_seen − delay. Set the delay from the measured distribution of received_at − event_time — near the p99, not the maximum — so the common case fires quickly. The gap between p99 and the max is deliberately handled by allowed lateness and side outputs, not by inflating the delay, because a delay large enough to catch the rarest straggler would add that latency to every window. A watermark is a heuristic, not a fact: it is sometimes wrong by design, which is exactly why allowed lateness exists.

Allowed lateness vs side outputs — what's the difference?

Allowed lateness and side outputs are two consecutive zones on the lateness timeline. Allowed lateness is a grace period, measured in event time past a window's end, during which the window retains its state and re-fires a corrected result whenever a late event arrives — the dashboard self-heals as stragglers land, provided the sink upserts on (window, key) so corrections replace rather than double-count. Side outputs (dead-letter channels) catch events that arrive later than even the grace period, once the window state has been purged and can no longer be corrected in-stream. The critical rule is that a past-grace event is never dropped silently: it is routed to an observable side output with full context, alerted on by volume, and folded into a nightly idempotent reconciliation. Allowed lateness gives real-time self-correction; side outputs guarantee eventual completeness with zero silent loss.

How is out-of-order data different from late data?

Out-of-order is about sequence; late is about the watermark. An event is out of order if it arrives after an event with a later event timestamp — a pure statement about arrival order, independent of any completeness marker. An event is late if it arrives after the watermark has already passed its window's end — a statement relative to the pipeline's completeness assertion. The two are related but not the same: every late event is necessarily out of order, but the vast majority of out-of-order events are not late, because the watermark delay is specifically sized to absorb the expected out-of-orderness. You handle out-of-orderness with the watermark delay and reorder buffers (a sequence problem, solved by buffering and sorting); you handle lateness with allowed lateness and side outputs (a watermark problem, solved by grace periods and dead-letters). Conflating them is the classic tell that a candidate has never run a real event-time job.

How do I guarantee exactly-once with reordering?

Exactly-once over an out-of-order, at-least-once source is a composition of three watermark-bounded patterns. First, idempotent keyed dedupe: maintain a set of seen event ids (or (key, sequence)) and skip anything already processed, committing the "seen" record and the side effect in one transaction so a crash cannot double-process or lose the event. Second, a bounded reorder buffer: a min-heap keyed by event time that releases events only once the watermark guarantees no earlier event can still arrive, producing ordered output. Third, sequence-gap detection on a per-key monotonic sequence so a genuinely lost record is detected (and alerted/reconciled) rather than silently skipped. The unifying discipline is that all of this state — dedupe keys, buffered events, pending sequences — is evicted at the watermark horizon plus allowed lateness, so the pipeline runs in memory proportional to the out-of-orderness bound rather than to total stream volume. That watermark-bounded eviction is what makes exactly-once achievable without unbounded state.

Practice on PipeCode

• Drill the streaming practice library → for the watermark, windowing, allowed-lateness, and exactly-once problems senior interviewers love.
• Rehearse on the event-processing practice library → for event-time semantics, reordering, side-output, and dead-letter reconciliation patterns.
• Sharpen the temporal axis with the time-series practice library → for sessionization, sliding-window aggregation, and sequence-gap detection scenarios.
• Stack the prerequisites against PipeCode's broader 450+ data-engineering catalogue to anchor the event-time, watermark, and lateness decision rules against real graded inputs.

Lock in late & out-of-order data muscle memory

Docs explain event time. PipeCode drills explain the decision — when processing time silently corrupts a count, how to size a watermark delay from measured skew, when a late event re-fires a window versus lands in a side output, and how to bound exactly-once state by the watermark. Pipecode.ai is Leetcode for Data Engineering — pattern-first practice tuned for the production trade-offs senior data engineers actually face.

Practice streaming problems →
Practice event-processing problems →