Change-Data-Capture Validation › Validating Debezium CDC Streams Against the Source

# Validating Debezium CDC Streams Against the Source

This page answers one concrete question: given a Debezium change stream flowing from a source database into Kafka, how do you prove — continuously and cheaply — that it is a complete, correctly ordered image of what the source committed? It sits under the change-data-capture validation reference and specializes its generic invariants for Debezium’s envelope, which exposes exactly the metadata the checks need: the source log position, the snapshot flag, and per-table heartbeats. This is the gate that runs before the watermark alignment stage trusts the stream.

# Problem Framing

You migrate an order-management PostgreSQL database to a new service and mirror every change through Debezium into Kafka for the duration of the cutover. Two silent failures keep migration engineers awake: a connector restart that resumes from a stored offset ahead of where it stopped, skipping a range of WAL, and a quiet lookup table that emits no changes for an hour, during which a real gap is indistinguishable from healthy silence. Debezium gives you the tools to close both: every change carries a monotonic source.lsn (or gtid/scn), the initial snapshot is fenced by snapshot markers, and heartbeat events advance a per-partition position even when a table is idle. The validator turns those three signals into a continuous completeness proof.

# Implementation

The validator parses the Debezium envelope, tracks the last accepted log position per partition, and treats heartbeats as position advances so an idle table cannot mask a gap. It anchors the first post-snapshot change to the snapshot’s last position.

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.debezium")


@dataclass(frozen=True)
class DebeziumEnvelope:
    partition: int
    lsn: int                 # source.lsn / gtid ordinal / scn
    op: Optional[str]        # "r" snapshot read, "c","u","d", or None for heartbeat
    snapshot: str            # "true", "last", or "false"
    is_heartbeat: bool = False


@dataclass
class DebeziumValidator:
    _last_lsn: Dict[int, int] = field(default_factory=dict)
    _snapshot_high: Optional[int] = None
    gaps: List[tuple] = field(default_factory=list)

    def observe(self, env: DebeziumEnvelope) -> None:
        """Advance per-partition position from any event, including heartbeats."""
        prev = self._last_lsn.get(env.partition)
        # Heartbeats and data events both advance the frontier; only data events gap-check.
        if not env.is_heartbeat and prev is not None and env.lsn > prev + 1:
            self.gaps.append((env.partition, prev, env.lsn))
            logger.error("Debezium gap on partition %d: %d -> %d", env.partition, prev, env.lsn)
        if env.snapshot == "last":
            self._snapshot_high = env.lsn
            logger.info("snapshot high-water mark = %d", env.lsn)
        self._last_lsn[env.partition] = max(prev, env.lsn) if prev is not None else env.lsn

    def anchor_first_change(self, first_stream_lsn: int) -> bool:
        """The first streamed change must directly continue the snapshot boundary."""
        if self._snapshot_high is None:
            logger.warning("no snapshot marker seen; cannot anchor")
            return False
        if first_stream_lsn <= self._snapshot_high:
            logger.warning("stream replays snapshot: %d <= %d", first_stream_lsn, self._snapshot_high)
            return False
        if first_stream_lsn > self._snapshot_high + 1:
            logger.error("gap at snapshot handoff: %d -> %d", self._snapshot_high, first_stream_lsn)
            return False
        return True

# Debezium Signal Coverage

Each envelope field closes a specific class of silent loss.

# Key Implementation Notes

• Heartbeats are not optional on quiet tables. Enable Debezium heartbeats so an idle partition still advances its position; without them, an hour of silence and an hour-long gap are indistinguishable, and the validator cannot assert completeness.
• Anchor on snapshot: last, not on the first change. The snapshot’s final marker is the only reliable boundary; the first streamed change must continue it by exactly one. This is the handoff most connector restarts corrupt.
• Track position per partition, then reduce. A Kafka topic is partitioned, so continuity holds per partition; a global check masks a gap that lives on one partition while others advance. Reconcile the union, as the reconciling Kafka topics with a source of truth runbook details.
• Cross-check slot lag out of band. A dropped replication slot causes a gap the stream itself cannot show; alert on source-side slot lag well below the WAL retention window so the connector never falls off the log.

# Verification

Assert gap detection, heartbeat masking prevention, and snapshot anchoring.

v = DebeziumValidator()
v.observe(DebeziumEnvelope(0, 100, "r", "true"))
v.observe(DebeziumEnvelope(0, 101, "r", "last"))       # snapshot high-water = 101
assert v.anchor_first_change(102) is True              # continues by exactly one
assert v.anchor_first_change(105) is False             # gap at handoff

v.observe(DebeziumEnvelope(0, 102, "c", "false"))
v.observe(DebeziumEnvelope(0, 105, "c", "false"))      # jumped 102 -> 105
assert v.gaps and v.gaps[-1] == (0, 102, 105)
logger.info("Debezium validation verified")

Run it as a Kafka consumer beside the reconciler and alert on any non-empty gaps list; a single detected gap must quarantine the affected key range and trigger a targeted incremental re-snapshot rather than allowing reconciliation to proceed on an incomplete stream.

# Operational Considerations

Consume the change and heartbeat topics with a dedicated least-privilege group so validation never competes with the reconciler for offsets, and size heartbeat cadence to the tightest gap-detection latency the migration requires — a one-second heartbeat on a critical table, a minute on a cold one. Keep per-partition state bounded and checkpoint it alongside the reconciler’s own checkpoint and state recovery so a validator restart resumes its continuity tracking rather than re-scanning. Persist every gap verdict with the source position range that produced it, because during a regulated cutover the ability to prove “no committed change was missed” is the sign-off criterion, and that proof is only as good as the retained position log.

# Related

• Change-data-capture validation — the parent reference whose invariants this specializes for Debezium.
• Reconciling Kafka topics with a source of truth — value-level reconciliation once the stream is proven complete.
• Watermark alignment strategies — the next stage, which trusts this gate’s completeness guarantee.
• Schema validation pre-checks — the analogous fail-fast contract gate on the batch path.