Watermark Alignment Strategies › Handling Late-Arriving Events in Reconciliation

Handling Late-Arriving Events in Reconciliation

This page answers a precise operational question: once a comparison window has been sealed by the low watermark, what do you do with a record that arrives anyway? It sits under the watermark alignment strategies reference and picks up where diagnosing phantom discrepancies in watermark-aligned streams leaves off: you have confirmed the stragglers are real, late, and frequent enough to matter, and now you need a policy that absorbs them without either dropping data or breaking the determinism that makes reconciliation auditable.

Problem Framing

A mobile analytics pipeline replays events buffered on devices that were offline for hours. Most records arrive within seconds of their event time; a small fraction arrive hours late. A five-minute lateness bound is right for the median but seals windows the stragglers will miss, and simply widening the bound to hours would delay every reconciliation verdict for the 99% that are on time. The correct design decouples the two: keep a tight watermark for timely sealing, and add a bounded allowed-lateness window during which a late record can still be routed to a side-output ledger and reconciled out of band — never silently discarded, and never at the cost of holding the main verdict hostage to the slowest device.

Implementation

The handler keeps a tight watermark for the primary comparison and a longer allowed-lateness horizon. Records inside the horizon but behind the watermark are appended to a side-output ledger keyed by window; records beyond the horizon are dropped only after being counted and logged, so “dropped” is always a measured, defensible decision.

python
import logging
from dataclasses import dataclass, field
from typing import Dict, List

logging.basicConfig(
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s", level=logging.INFO
)
logger = logging.getLogger("recon.streaming.late")


@dataclass(frozen=True)
class Event:
    key: str
    event_time_ms: int
    window_id: int


@dataclass
class LateEventRouter:
    allowed_lateness_ms: int
    _side_output: Dict[int, List[Event]] = field(default_factory=dict)
    dropped_total: int = 0

    def route(self, event: Event, low_watermark_ms: int) -> str:
        """Route a record to on-time, late-but-recoverable, or dropped (counted)."""
        if event.event_time_ms > low_watermark_ms:
            return "on_time"                       # still inside an open window
        lateness = low_watermark_ms - event.event_time_ms
        if lateness <= self.allowed_lateness_ms:
            self._side_output.setdefault(event.window_id, []).append(event)
            logger.info("late-but-recoverable key=%s lateness=%dms", event.key, lateness)
            return "side_output"
        self.dropped_total += 1
        logger.warning(
            "dropped key=%s lateness=%dms beyond allowed %dms",
            event.key, lateness, self.allowed_lateness_ms,
        )
        return "dropped"

    def drain_window(self, window_id: int) -> List[Event]:
        """Return late records for a window, in deterministic order, to re-reconcile."""
        events = self._side_output.pop(window_id, [])
        return sorted(events, key=lambda e: (e.event_time_ms, e.key))

Re-reconciling a sealed window is deterministic because the side-output drain sorts by event time and key before replay: the same set of late events always produces the same corrected verdict, so a window reopened twice yields identical results.

Late-Event Policy Trade-Off

The right horizon and disposition come from how costly a missed record is versus how long the verdict can wait.

Policy Verdict latency Completeness State cost Compliance / regulatory
Tight watermark, no lateness Lowest Loses every straggler Minimal Weak: silent loss is indefensible in an audit.
Tight watermark + side-output ledger Low for the verdict; late records reconciled out of band High; stragglers recovered within the horizon Bounded by horizon × late rate Strong: every late record is ledgered and its disposition recorded.
Wide watermark High for every verdict High High buffer for all events Moderate: complete but violates latency objectives.
Reopen-and-correct Low, then corrected Highest Requires retained window state Strongest: the correction and its trigger are both auditable.

Key Implementation Notes

  • Decouple sealing from completeness. The watermark should stay tight so verdicts are timely; allowed-lateness is a separate horizon that governs recovery. Conflating them forces a false choice between latency and correctness.
  • Never drop silently — count, log, then drop. A record beyond the allowed-lateness horizon may legitimately be dropped, but only as a measured event with a metric and a log line, so “we discard events later than N” is a stated policy an auditor can review, not an accident.
  • Determinism survives reopening only if replay is ordered. Sort the side-output by event time and key before re-reconciling; an unordered replay can produce a different corrected verdict on re-run and destroys the audit trail.
  • Bound the ledger with the horizon. Side-output state is O(late events within the horizon); evict a window’s ledger once its allowed-lateness has fully elapsed, or the straggler buffer grows without limit on a bursty source.

Verification

Assert the three routing outcomes and that a reopened window replays deterministically.

python
r = LateEventRouter(allowed_lateness_ms=60_000)
wm = 500_000
assert r.route(Event("a", 600_000, 1), wm) == "on_time"
assert r.route(Event("b", 460_000, 1), wm) == "side_output"   # 40s late, within horizon
assert r.route(Event("c", 300_000, 1), wm) == "dropped"       # 200s late, beyond horizon
assert r.dropped_total == 1
# Deterministic replay order regardless of insertion order.
r.route(Event("e", 470_000, 2), wm)
r.route(Event("d", 465_000, 2), wm)
assert [e.key for e in r.drain_window(2)] == ["d", "e"]
logger.info("late-event routing verified")

Operational Considerations

Expose late_events_side_output_total, late_events_dropped_total, and the lateness distribution so the allowed-lateness horizon is set from data, not intuition — target a horizon that captures the P99.9 of observed lateness while dropping only the genuinely pathological tail. Run the side-output reconciliation on a slower cadence than the primary pass so it never competes for the resources timely verdicts need, and persist each reopened-window correction with its trigger and the late events that caused it. Because a late record carries the same regulated fields as an on-time one, the side-output ledger is an in-scope data store: apply the same retention and masking policy defined in the security boundaries for reconciliation reference.