Streaming & CDC Reconciliation › Change-Data-Capture Validation

Change-Data-Capture Validation

Change-data-capture validation is the gate that must pass before any streaming comparison is trusted: the workload that proves the change stream itself is a complete, ordered, gap-free image of what the source actually committed. Reconciling a target against a stream that has already silently dropped a change is worse than not reconciling at all, because it launders corruption through a green dashboard. This reference sits inside the streaming reconciliation pipelines stage and specializes it for source-of-truth integrity, written for the data engineers, migration specialists, Python pipeline builders, and platform operations teams who run Debezium, native logical replication, or a cloud CDC service into a log the rest of the pipeline consumes on faith.

The failure this reference prevents is subtle because CDC connectors are engineered to look healthy while losing data. A connector restart that resumes from the wrong log-sequence number skips a range of commits; a topic compaction removes an intermediate update; an at-least-once delivery replays a change so a downstream apply runs twice. None of these surface as an error — the stream keeps flowing — and none are visible to a snapshot comparison taken after both sides re-converge. The only defense is to validate the stream against invariants the source’s transaction log guarantees: monotonic sequence continuity, operation-type coherence per key, and a clean handoff from the initial snapshot to live change capture. This gate runs before the watermark alignment stage, because aligning a stream that is already missing records only produces confident wrong answers.

CDC validation invariants between the source log and the change topic A source transaction log emits commits with monotonically increasing sequence numbers into a connector, which writes them to a change topic. Three validation checks tap the topic: sequence continuity confirms no gap between consecutive numbers, operation coherence confirms no update or delete precedes an insert for a key, and snapshot anchoring confirms the first streamed sequence continues the initial snapshot's high-water mark. A verdict gate routes the stream to trusted, or to a gap-and-quarantine path when any invariant fails. Source txn log LSN 100,101,102… Connector Debezium Change topic Kafka / Kinesis Validation checks continuity · coherence · anchor Snapshot high-water mark Trusted stream → watermark alignment Gap detected quarantine & re-snapshot
Three invariants tap the change topic — sequence continuity, per-key operation coherence, and snapshot anchoring — and a verdict gate either declares the stream trusted for alignment or routes a detected gap to quarantine and a targeted re-snapshot.

Architectural Boundaries

This workload consumes the raw change topic and the source’s replication metadata — log-sequence numbers, the initial snapshot’s completion offset, and per-key primary-key values. It produces a single trust verdict per offset range plus a structured gap report when an invariant fails. It does not compare source and target values — that is the comparator’s job downstream of watermark alignment. It only answers one question: is this stream a faithful, complete image of what the source committed? Everything else in the streaming stage assumes a yes from this gate, so a false positive here is the most expensive defect in the pipeline.

Prerequisites

Step-by-Step Implementation

1. Verify log-sequence continuity

A gap is any jump greater than one between consecutive per-partition sequence numbers after ordering. Track the last accepted sequence per partition and flag discontinuities; a real gap means committed source changes never reached the topic.

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

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


@dataclass(frozen=True)
class CdcEvent:
    key: str
    op: str                # "r" snapshot, "c" insert, "u" update, "d" delete
    lsn: int               # source log-sequence number
    partition: int
    ts_ms: int


@dataclass
class ContinuityReport:
    gaps: List[tuple] = field(default_factory=list)          # (partition, after_lsn, before_lsn)
    duplicates: List[tuple] = field(default_factory=list)    # (partition, lsn)
    is_continuous: bool = True


def verify_continuity(events: List[CdcEvent]) -> ContinuityReport:
    """Detect missing or replayed log-sequence numbers per partition."""
    report = ContinuityReport()
    last: Dict[int, int] = {}
    for e in sorted(events, key=lambda x: (x.partition, x.lsn)):
        prev = last.get(e.partition)
        if prev is not None:
            if e.lsn == prev:
                report.duplicates.append((e.partition, e.lsn))
            elif e.lsn > prev + 1:
                report.gaps.append((e.partition, prev, e.lsn))
        last[e.partition] = max(prev, e.lsn) if prev is not None else e.lsn
    report.is_continuous = not report.gaps
    if report.gaps:
        logger.error("CDC continuity broken: %d gap(s) detected", len(report.gaps))
    return report

2. Check per-key operation coherence

The lifecycle of a key is constrained: an update or delete before any insert, or an operation after a delete without a re-insert, signals a lost event even when sequence numbers look continuous because the missing change was compacted or filtered.

python
def verify_operation_coherence(events: List[CdcEvent]) -> List[str]:
    """Return keys whose operation sequence is impossible, implying a lost change."""
    state: Dict[str, str] = {}          # key -> last op
    violations: List[str] = []
    for e in sorted(events, key=lambda x: (x.ts_ms, x.lsn)):
        prev = state.get(e.key)
        if e.op in ("u", "d") and prev in (None, "d"):
            # update/delete with no live row implies a missing insert.
            violations.append(e.key)
        state[e.key] = e.op
    if violations:
        logger.error("operation coherence violated for %d key(s)", len(violations))
    return violations

3. Anchor the stream to the initial snapshot

The most common silent gap is at the snapshot-to-stream boundary. Assert that the first streamed change continues the sequence where the consistent snapshot ended — no lower (a replay) and no higher with a gap (a loss). This handoff is the anchor the whole reconciliation of Kafka topics against a source of truth depends on.

python
def anchor_to_snapshot(snapshot_high_lsn: int, first_stream_lsn: int, max_gap: int = 1) -> bool:
    """The first live change must continue the snapshot's high-water mark without a gap."""
    if first_stream_lsn <= snapshot_high_lsn:
        logger.warning("stream replays snapshot range: %d <= %d", first_stream_lsn, snapshot_high_lsn)
        return False
    if first_stream_lsn > snapshot_high_lsn + max_gap:
        logger.error("gap at snapshot handoff: %d -> %d", snapshot_high_lsn, first_stream_lsn)
        return False
    return True

Validation Strategy Trade-Off

Layer validators by cost: run the cheap structural checks continuously and reserve value-level reconciliation for anchored intervals.

Strategy Latency Coverage of loss Cost Compliance / regulatory
Sequence continuity Continuous, per event Catches dropped and replayed ranges Negligible; integer arithmetic Strong: the source LSN is an authoritative, signable anchor.
Operation coherence Continuous, per key Catches compaction and filter loss continuity misses Low; O(events) with per-key state Strong: reconstructs the committed lifecycle from the log.
Count reconciliation Per interval Catches net cardinality drift, not offsetting errors Moderate; periodic aggregation Moderate: proves totals, not per-row fidelity.
Full-row hash reconciliation Per interval / on demand Catches value-level divergence High; reads payloads on both sides Strongest: pair with a NIST-approved hash for an auditable per-row proof.

Scaling and Performance

Continuity and coherence checks are streaming-friendly: both hold O(active keys) state and run in a single pass, so shard them by key or partition and combine per-shard reports with a simple merge. The expensive validators — count and full-row-hash reconciliation — should run only on offset ranges the cheap checks have already anchored, so a full read is never wasted on a stream already known to have a gap. Keep the per-key state in a bounded structure with TTL eviction keyed on the tolerated-lateness window, or the coherence tracker becomes an unbounded map on a high-churn table. When the connector is horizontally scaled, validate per connector task and reconcile the union, because a gap on one task is invisible in another task’s continuous sequence.

Failure Modes and Diagnostic Runbook

  • Snapshot-handoff gap. Cause: the connector began streaming from a stored offset ahead of the snapshot’s high-water mark after a restart. Detection signal: anchor_to_snapshot fails at task start. Remediation: trigger a targeted incremental re-snapshot of the affected key range and replay from the snapshot boundary.
  • Silent compaction loss. Cause: log or topic compaction removed an intermediate update. Detection signal: operation-coherence violations without any continuity gap. Remediation: disable compaction on reconciled topics, or reconcile against the source at value level for the affected keys.
  • Duplicate application. Cause: at-least-once redelivery after a consumer restart. Detection signal: duplicate LSNs in the continuity report. Remediation: make the downstream apply idempotent per exactly-once reconciliation with idempotent sinks.
  • Replication-slot lag stall. Cause: the source retained WAL faster than the connector consumed it, and the slot was dropped. Detection signal: a sequence gap coinciding with a slot-lag alert. Remediation: re-snapshot; then bound slot lag with an alert threshold well below the retention window.

Deep Dives