Handle Late and Out-of-Order Events in Stream Processing

Problem and Scope

Mobile clients report purchase events with at least eventid, eventtime, userid, and an amount already converted to one currency. Network retries can create duplicates, offline devices can upload several hours later, and different partitions do not preserve a global order. Compute revenue for each UTC clock hour by eventtime, with a peak input of 20,000 events per second.

The business wants an estimate for the current hour within one minute, an on-time result after the watermark passes the end of the window, and automatic corrections for 24 hours after that window ends. Data arriving beyond 24 hours must not disappear silently; it goes to audit and offline reconciliation. Throughput, timeliness, and the correction period are problem inputs, not performance claims about a stream-processing engine.

Assume the producer assigns a stable event_id, and valid retries with that ID carry the same business content. Raw events remain in replayable storage. The scope covers time semantics, watermarks, window triggers, deduplication, result updates, state capacity, recovery, and reconciliation. Broker selection and currency conversion are out of scope. This is a data-engineering question: the goal is a verifiable data contract that makes the completeness, latency, and state-cost tradeoff explicit.

What Interviewers Evaluate

The first signal is whether the candidate separates three clocks. eventtime determines the business hour containing an event. processingtime says when the engine sees it and can drive a one-minute early refresh. A watermark is the engine's estimate of event-time progress. If processing time determines the window, replaying the same history at a different time can produce different hourly results.

The second signal is treating a watermark as a progress estimate rather than an absolute promise. Advancing it quickly gives timely output but classifies more records as late. Holding it back improves completeness but delays window closure and retains more state. A strong answer derives the policy from measured arrival-delay data and the correction SLO instead of memorizing a fixed five-minute or one-hour delay.

The third signal is separating the watermark, allowed lateness, and deduplication retention. The watermark controls the on-time firing. Allowed lateness controls how long window state remains correctable. Deduplication state must span the period in which another copy of an event can arrive. The values may be related, but they are not one setting.

Finally, the output has to converge. Late events cause multiple results for one window. Appending every snapshot makes a downstream sum count the window repeatedly. Results need an idempotent upsert keyed by the window, with a revision or finality marker. A checkpoint protects operator state; if the sink cannot commit with the checkpoint, the design still needs transactional writes or idempotent version replacement.

Questions to Clarify Before Answering

• Is the business metric based on occurrence or arrival? This problem uses purchase event_time. Processing time is appropriate only when the metric is specifically “requests processed by the system now.”
• Is the one-minute result an estimate or a final number? It is an estimate here, so a processing-time early trigger is valid. If every displayed result must be complete, the system has to wait longer or use batch computation.
• Can reports change after 24 hours? The 24 hours are the streaming job's automatic correction period. Later data goes to reconciliation. A regulatory or settlement requirement to correct every late record means state cleanup cannot define the data's ultimate truth.
• Can duplicates be identified by ID? A stable event_id exists here. Guessing from user, amount, and time will both remove legitimate purchases and miss duplicates; fix the producer contract first.
• What if one ID carries different content? Do not choose first-write-wins or last-write-wins. Record a payload-fingerprint conflict and quarantine it because the producer violated the idempotency contract.
• Does the sink accept snapshots, deltas, or retractions? This design emits complete window snapshots and upserts by (window_start, dimensions). An append-only sink needs a versioned changelog whose read layer selects the latest revision.
• Can the event timestamp be trusted? Quarantine timestamps far in the future, older than raw-data retention, or invalid for the agreed timezone. A bad future timestamp participating in maximum event time can advance the watermark too aggressively.

30-Second Answer

“I would assign UTC hourly windows by eventtime, generate a watermark per active source partition, and advance at the minimum safe progress. A processing-time trigger emits estimates each minute; the watermark emits the on-time result, with state retained for 24 hours of corrections. eventid and a payload fingerprint deduplicate events, while the sink upserts by window and revision. Too-late data goes to a side output and daily reconciliation. Recovery combines a replayable source, checkpoints, and a transactional or idempotent sink.”

Step-by-Step Deep Dive

Start with the result contract. Use half-open windows such as [10:00, 11:00), and include at least window_start and the reporting dimensions in the key. Every output is the complete revenue snapshot for that window at its current version:

HourlyRevenue {
  window_start
  window_end
  revenue
  revision
  result_state  // EARLY | ON_TIME | FINAL
}

revision increases monotonically for a window, and the sink accepts only a larger revision. EARLY is the one-minute refresh and does not claim completeness. ON_TIME means the watermark has passed the window end. FINAL means the 24-hour automatic correction period has ended. FINAL is an operational contract, not a claim that no more data exists; reconciliation still handles records beyond the cutoff.

Derive the time policy from the constraints. Each active source partition extracts a validated event_time. A common bounded-out-of-orderness strategy is:

partition_watermark = max_valid_event_time_seen - out_of_order_bound
operator_watermark = min(active_partition_watermarks)

Choose outoforderbound from an observed arrival-delay distribution, the acceptable late-data rate, and the ONTIME latency target. Per-partition watermarks prevent a fast partition from declaring a slow one complete. A multi-input operator takes the minimum progress so that it does not pass an input that can still produce older records. Mark a partition idle only after it has produced no data for an agreed interval; otherwise it can stall the global watermark forever. An overly short idleness threshold makes old records from a resumed partition late, so the side output remains necessary.

The window has two trigger families. A processing-time early trigger emits the current complete snapshot once a minute. When the watermark passes windowend, emit the ONTIME revision. Retain window state until the watermark passes window_end + 24h; each valid late event updates the aggregate and produces a higher revision. At cleanup, emit FINAL and release the state. Beam's trigger and accumulation model illustrates why “when to emit” and “whether a pane is a delta or accumulated snapshot” are separate choices. This design uses accumulated snapshots and upserts, so downstream systems do not have to assemble panes.

Deduplicate by eventid before aggregation. Store eventid → (eventtime, payloadfingerprint). Let the first event through, discard another copy with the same fingerprint, and send the same ID with a different fingerprint to a conflict stream. Retention depends on the longest interval in which the upstream can replay another copy, not merely on the window size. Spark's watermark-based deduplication semantics likewise require a delay threshold longer than the timestamp gap between the earliest and latest duplicate. State cleanup that happens too soon lets a late duplicate count again.

Estimate state explicitly. If the peak of 20,000 events per second lasts for 24 hours and every event is unique, the job remembers 20,000 × 86,400 = 1,728,000,000 IDs. Even an illustrative logical payload of only 40 bytes per entry makes this upper bound about 69.12 GB. Engine objects, indexes, the state backend, checkpoints, and replicas increase the physical footprint. State therefore needs key-based partitioning and incremental checkpoints, while retention follows the replay contract. If 99% of duplicates arrive in a shorter period, a shorter streaming horizon can be combined with a sink business key and offline reconciliation, but the residual duplicate risk must be quantified rather than hidden behind a TTL.

The output path must tolerate replay. Ideally, the sink participates in checkpoint transactions so that source position, operator state, and window output commit together. Otherwise, make (window_key, revision) an idempotent upsert: replaying the same or an older revision after a crash cannot overwrite the new value. A broker's exactly-once label is insufficient when an external database write sits outside checkpoint confirmation. End-to-end semantics depend on the shared commit boundary of source, state, and sink.

Events beyond 24 hours go to a too-late side output and remain in the raw log. A daily batch job recomputes affected windows by event_time with the same ID and amount rules, then compares them with the streaming FINAL snapshots. A material difference creates a higher revision or enters financial approval. The batch job atomically replaces a window or performs a versioned upsert. Adding the batch result to the existing total would count the same data twice when the batch is replayed.

Use a minimal sequence to verify the semantics. Window [10:00, 11:00) receives A=100, B=50, and then an identical copy of A. The deduplicated ON_TIME result is 150. After the watermark passes 11:00, C=20 arrives during allowed lateness, and a higher revision changes the window to 170. D arrives after the watermark passes the cleanup point and goes to reconciliation instead of directly touching cleared state. If A's second payload has amount 120, it goes to the conflict stream; the result must not become 170 or 190.

The test matrix also covers disorder across partitions, an idle partition that stalls the watermark, quarantine of a future timestamp, boundaries immediately before and after the watermark, crashes before and after the sink write, checkpoint recovery and replay, out-of-order revisions reaching the sink, and repeated reconciliation runs. Production metrics include p50/p95/p99 and tail arrival delay, current watermark and lag per operator, early/on-time/late/too-late counts and amounts, deduplication hits and fingerprint conflicts, state bytes, checkpoint duration, rejected sink revisions, and batch-versus-stream deltas.

Strong Sample Answer

“I would define the one-minute number as an estimate and the 24 hours as the automatic correction period. event_time assigns a purchase to a UTC hour; processing time only drives the one-minute early trigger. Each active input partition generates a watermark from its largest validated event time minus the disorder allowance, and downstream uses the slowest active input. Idle partitions are marked explicitly, and invalid future timestamps are quarantined.

Before aggregation, I deduplicate by eventid, storing event time and a payload fingerprint. An identical retry counts once; the same ID with different content enters a conflict stream. The natural-hour window emits a complete EARLY snapshot each minute, an ONTIME snapshot after its end is behind the watermark, and retains state for 24 hours. Late events in that period update the total and increase the revision; cleanup emits FINAL.

The sink upserts by window key and revision instead of adding every snapshot as a delta. The source is replayable and operator state restores from checkpoints. If the sink cannot join a checkpoint transaction, revisioned writes prevent recovery replay from replacing a newer result. Events later than 24 hours go to a side output, and a daily batch job recomputes the raw log and compares it with FINAL.

If the peak of 20,000 events per second lasts for 24 hours and all IDs are unique, the upper bound is 1.728 billion deduplication entries. Even 40 logical bytes per entry is 69.12 GB, so measured delay distributions must justify retention, and state and checkpoint cost need monitoring. I would finish with fault injection for duplicates, disorder, lateness, future timestamps, idle partitions, and recovery, asserting that the final window equals the offline recomputation.”

Common Mistakes

• Assign business windows by processing time → the same event can land in a different hour during a historical replay → use event time for membership and processing time only for early output.
• Describe the watermark as “no older data can arrive” → it is usually a heuristic progress estimate, so an older event can still appear → define allowed lateness, a too-late side output, and reconciliation.
• Use one unexplained value for watermark delay, allowed lateness, and deduplication TTL → they govern firing, window state, and duplicate recognition respectively → derive them from the latency SLO, correction period, and upstream replay contract.
• Append a total on every firing → EARLY, ON_TIME, and late firings are summed repeatedly → emit a complete snapshot and upsert by window and revision, or define a retractable-delta protocol.
• Advance from a global maximum event time → a fast partition or bad future timestamp makes slow-partition data late too early → generate partition progress, take the active-input minimum, and validate timestamps.
• Keep a permanently quiet partition in the minimum → the watermark stops, preventing windows and deduplication state from being cleaned → use observable idleness handling and route resumed late data correctly.
• Store only the event ID → producer reuse of an ID is silently swallowed → also store a payload fingerprint and quarantine conflicts.
• Say “exactly once is enabled” → an external sink may not share the checkpoint boundary → trace the end-to-end commit path and use a transactional sink or versioned idempotent writes.
• Drop data after state cleanup → the stream looks stable while financial results become inexplicable → retain raw events and implement side output, batch recomputation, and discrepancy review.

Follow-up Questions and Answers

Follow-up 1: How large should the watermark's disorder allowance be?

Measure processingtime - eventtime in production and segment it by source, client version, and region. First choose the latest acceptable ON_TIME result and the share of events allowed into the correction path; then select a percentile that satisfies both. Continue monitoring p95, p99, and the tail. A distribution change should go through a configuration release rather than letting one outlier move the policy automatically. The watermark governs on-time firing, while the 24-hour correction period still covers a longer tail.

Follow-up 2: Why can one quiet Kafka partition stop the window?

The safe progress of a multi-input operator is the minimum input watermark. A partition that does not advance keeps that minimum unchanged. After confirming it has produced no events for the idleness threshold, mark it idle so it temporarily leaves the minimum. Do not set the threshold too short: records from a resumed partition can be older than the new watermark. They must enter allowed-late processing or the side output, and idle-state transitions need their own metric.

Follow-up 3: Why does 24-hour deduplication state not promise that duplication is impossible forever?

Twenty-four hours covers only this problem's automatic correction period. If the upstream replays an ID in hour 25, the streaming state is gone and the event can re-enter the aggregate. Permanent deduplication needs a longer-lived business unique index, a queryable event registry, or full offline deduplication; each adds storage and write cost. State the guarantee as “deduplicated within the declared replay horizon,” then reconcile later records.

Follow-up 4: What if the downstream database supports INSERT but not upsert?

Write every result to an immutable changelog keyed by window and revision. The read layer builds the current view from the maximum revision for each window. Consumers must know that each record is a complete snapshot, not a delta, and must not sum all revisions. If query cost is too high, compact asynchronously into a serving table that supports atomic replacement while retaining the version log for audit and recovery.

Follow-up 5: How do you prove recovery neither overcounts nor undercounts?

For a fixed input, record the expected deduplication set and window revisions. Terminate the job at three boundaries: after the source read but before a state checkpoint, after state checkpointing but before sink confirmation, and after the sink write but before checkpoint confirmation. Replay after recovery and assert that each event ID contributes once, the sink retains only the largest revision, and the final window equals an offline recomputation. A RUNNING status without fault injection at these commit boundaries is not proof of end-to-end correctness.