Streaming & CDC Reconciliation › Watermark Alignment Strategies

Watermark Alignment Strategies

Watermark alignment is the timing discipline that makes streaming reconciliation possible: the mechanism that decides, for two continuously written systems, the exact moment at which a slice of event time is sealed and safe to compare. Without it, every comparison races in-flight records and reports differences that are nothing but latency. This reference sits inside the streaming reconciliation pipelines stage and specializes it for the single hardest problem in the stage — turning a reordered, multi-partition change stream into a monotonic frontier past which the source and target can be declared provably equal. It is written for the data engineers, migration specialists, Python pipeline builders, and platform operations teams who own continuous replication and must distinguish a real dropped record from a merely late one.

A watermark is a claim, not a clock. When a reconciler advances its low watermark to time W, it asserts that no event with an event time at or below W will ever arrive again on any partition. Everything downstream — the windowed comparison, the checkpoint commit, the discrepancy verdict — depends on that claim being conservative: set the watermark too aggressively and a late record arrives after its window closed, producing a phantom discrepancy; set it too loosely and reconciliation lags reality by more than the business can tolerate. This reference defines how to make that trade deliberately rather than by accident.

Per-partition frontiers combine into a conservative low watermark Three partition lanes each show events plotted by event time, with each lane's rightmost event marking its frontier. Partition two lags furthest behind. The low watermark is drawn as a vertical line at the slowest partition's frontier minus a tolerated-lateness margin. Events to the left of the watermark are sealed and released for comparison; events to the right remain buffered. One late event sits just left of a partition frontier but right of the watermark, showing the margin that protects it. event time → sealed · released for comparison buffered · not yet safe low watermark partition 0 partition 1 partition 2 frontier slowest frontier tolerated lateness
The low watermark trails the slowest partition's frontier by the tolerated-lateness margin; everything to its left is sealed and released for comparison, everything to its right stays buffered until the frontier advances.

Architectural Boundaries

This workload begins where the change-data-capture validation stage hands off a validated, per-partition stream of change events, each carrying a source commit timestamp and a monotonic offset. It ends when it emits, in event-time order, the batches of events a window can safely compare, plus the current low watermark that the checkpoint and state recovery stage will persist. It consumes ordering within a partition and produces ordering across partitions; it never reads target state or renders a parity verdict — that is the comparator’s job. Keeping the aligner free of comparison logic is what makes it deterministically testable in isolation.

Prerequisites

Step-by-Step Implementation

1. Track a per-partition frontier

The frontier of a partition is the maximum event time seen on it. The global low watermark is the minimum frontier across all active partitions, minus the tolerated-lateness margin — a conservative bound that no in-order partition can violate.

python
import logging
import time
from dataclasses import dataclass, field
from typing import Dict, Iterator, List, Optional

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


@dataclass(frozen=True)
class ChangeEvent:
    key: str
    op: str
    event_time_ms: int
    partition: int
    offset: int
    after: Optional[dict] = None


@dataclass
class Frontier:
    """Tracks the max event time and last-update wall clock per partition."""
    event_time_ms: int
    updated_wall_ms: int

2. Advance a bounded-out-of-orderness watermark

Bounded-out-of-orderness is the workhorse strategy: assume events are late by no more than a fixed bound, and place the watermark exactly that far behind the slowest frontier. Idle partitions are aged out so one silent shard cannot stall the global frontier.

python
@dataclass
class BoundedWatermarkAligner:
    max_lateness_ms: int
    idle_timeout_ms: int = 30_000
    _frontiers: Dict[int, Frontier] = field(default_factory=dict)
    _buffer: List[ChangeEvent] = field(default_factory=list)

    def _now_ms(self) -> int:
        return int(time.time() * 1000)

    def low_watermark(self) -> int:
        now = self._now_ms()
        active = [
            f.event_time_ms
            for f in self._frontiers.values()
            if now - f.updated_wall_ms <= self.idle_timeout_ms
        ]
        if not active:
            return -1
        return min(active) - self.max_lateness_ms

    def push(self, event: ChangeEvent) -> Iterator[ChangeEvent]:
        prev = self._frontiers.get(event.partition)
        new_time = max(prev.event_time_ms, event.event_time_ms) if prev else event.event_time_ms
        self._frontiers[event.partition] = Frontier(new_time, self._now_ms())
        self._buffer.append(event)
        yield from self._drain(self.low_watermark())

    def _drain(self, wm: int) -> Iterator[ChangeEvent]:
        if wm < 0:
            return
        ready = [e for e in self._buffer if e.event_time_ms <= wm]
        self._buffer = [e for e in self._buffer if e.event_time_ms > wm]
        # Deterministic release order: event time, then partition, then offset.
        for e in sorted(ready, key=lambda x: (x.event_time_ms, x.partition, x.offset)):
            yield e

3. Release sealed events into the comparator

The aligner’s only output is an ordered stream of sealed events plus the watermark that sealed them. Anything older than the watermark that still arrives is late and is routed to the late-event path rather than silently dropped — the subject of handling late-arriving events in reconciliation.

python
def align(source: Iterator[ChangeEvent], aligner: BoundedWatermarkAligner) -> Iterator[ChangeEvent]:
    """Yield sealed events in event-time order; log any record already behind the watermark."""
    for event in source:
        wm = aligner.low_watermark()
        if wm >= 0 and event.event_time_ms <= wm:
            logger.warning(
                "late event key=%s event_time=%d behind watermark=%d",
                event.key, event.event_time_ms, wm,
            )
            # Late records are handed to the late-arrival policy, never dropped here.
            continue
        yield from aligner.push(event)

Watermark Strategy Trade-Off

Choose the strategy from the correctness the reconciliation must guarantee and the lateness the source actually exhibits — not from a default.

Strategy Latency to seal Correctness under reorder Memory Compliance / regulatory
Bounded-out-of-orderness One lateness bound behind the frontier Correct if real lateness ≤ bound; late records escape past it Bounded by lateness window × arrival rate Auditable: the bound is a single declared parameter recorded with every verdict.
Per-partition punctuated As fast as the punctuation cadence Exact when the source emits explicit watermark markers Low; no time-based buffer Strongest: the source itself asserts completeness, giving a signed frontier.
Percentile / heuristic Adaptive, tracks observed lateness Probabilistic; a fixed fraction of events will be late by design Bounded by the sampling window Weakest: a statistical frontier is hard to defend in an audit; pair with a late-event ledger.
Processing-time fallback Lowest None under reorder — replays differ Minimal Unusable for regulated parity; acceptable only for coarse liveness alerts.

Scaling and Performance

Alignment state is O(events within the lateness window), so the two levers that bound memory are the tolerated-lateness value and the arrival rate. Keep the buffer as a per-partition structure so draining is a partial sort rather than a full re-sort of the global buffer, and shard the aligner by key range so no single instance holds the whole stream’s in-flight set. On high-cardinality streams, prefer a heap keyed on event time over repeated list comprehensions to keep the drain step at O(log n) per release. The watermark computation itself is O(partitions) and runs on every push, so cache the current minimum frontier and recompute only when the slowest partition advances. Across a horizontally scaled reconciler, watermarks must be combined by taking the global minimum of per-instance watermarks — a reduce, never an average — so one lagging shard correctly holds the global frontier back.

Failure Modes and Diagnostic Runbook

  • Stalled watermark. Cause: an idle or dead partition whose frontier never advances. Detection signal: watermark_lag_ms climbs without bound while throughput is healthy. Remediation: confirm the idle-partition timeout is set and firing; if a partition is genuinely dead, exclude it explicitly rather than letting it freeze the frontier.
  • Watermark overrun. Cause: a shard with a fast clock drags the minimum frontier ahead of true event time. Detection signal: a rising rate of events arriving already behind the watermark. Remediation: enable clock-skew normalization; widen the lateness bound only as a last resort, because it inflates latency for every partition.
  • Phantom discrepancies. Cause: the lateness bound is smaller than real-world lateness, so records land after their window closed. Detection signal: discrepancies that resolve themselves on the next full pass. Remediation: follow diagnosing phantom discrepancies in watermark-aligned streams.
  • Unbounded buffer growth. Cause: the lateness window is set far larger than needed, or a partition stopped draining. Detection signal: events_buffered grows linearly with time. Remediation: tighten the lateness bound to the observed P99 lateness and confirm every partition is advancing.

Deep Dives