Change-Data-Capture Validation › Reconciling Kafka Topics with a Source of Truth

Reconciling Kafka Topics with a Source of Truth

This page answers a specific question: once continuity checks prove a Kafka change topic has no structural gaps, how do you prove its contents match the source of truth at the value level? It sits under the change-data-capture validation reference and moves from “no events are missing” to “the events say the right thing.” The distinction matters because a topic can be perfectly continuous and still diverge: a transformation in the connector, a compaction that changed which version of a key survives, or a producer bug that wrote a stale value. Continuity is necessary; value reconciliation is what proves fidelity.

Problem Framing

A compacted Kafka topic is the materialized current state of a source table — one record per key, latest value wins. You must confirm that reading the whole compacted topic reproduces the source table exactly: same key set, same latest value per key, with deletes represented as tombstones that correctly remove keys. Two Kafka-specific hazards complicate this. Compaction is asynchronous, so a topic may transiently hold superseded versions that a naive comparison flags as divergence. And a partition reassignment can reorder consumption so a per-key “latest” is computed wrong unless you resolve latest by source offset, not by consumption order. The reconciler must be robust to both.

Implementation

The reconciler folds the compacted topic into a per-key latest-value map resolved by source offset, hashes each surviving value with a deterministic serializer, and compares the hash set against the source. Tombstones remove keys so a deleted row is proven absent, not merely stale.

python
import hashlib
import json
import logging
from dataclasses import dataclass
from typing import Dict, Iterable, Optional, Tuple

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


@dataclass(frozen=True)
class TopicRecord:
    key: str
    value: Optional[dict]     # None == tombstone (delete)
    source_offset: int        # source LSN, the authority for "latest"


def canonical_hash(value: dict) -> str:
    """Deterministic per-value fingerprint, stable across engines and encodings."""
    payload = json.dumps(value, sort_keys=True, separators=(",", ":"), default=str)
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()


def fold_compacted(records: Iterable[TopicRecord]) -> Dict[str, str]:
    """Resolve latest value per key by source offset; drop tombstoned keys."""
    latest: Dict[str, Tuple[int, Optional[dict]]] = {}
    for r in records:
        prev = latest.get(r.key)
        if prev is None or r.source_offset >= prev[0]:
            latest[r.key] = (r.source_offset, r.value)
    # Surviving keys hash their value; tombstoned keys are excluded entirely.
    return {k: canonical_hash(v) for k, (_, v) in latest.items() if v is not None}


def reconcile(topic_state: Dict[str, str], source_state: Dict[str, str]) -> Dict[str, list]:
    """Compare per-key hashes; return missing, extra, and mismatched keys."""
    report = {
        "missing_in_topic": sorted(k for k in source_state if k not in topic_state),
        "extra_in_topic": sorted(k for k in topic_state if k not in source_state),
        "value_mismatch": sorted(
            k for k in source_state.keys() & topic_state.keys()
            if source_state[k] != topic_state[k]
        ),
    }
    total = sum(len(v) for v in report.values())
    (logger.error if total else logger.info)(
        "kafka reconcile: %d missing, %d extra, %d mismatch",
        len(report["missing_in_topic"]), len(report["extra_in_topic"]),
        len(report["value_mismatch"]),
    )
    return report

Reconciliation Strategy Trade-Off

Layer cheap aggregate checks continuously and reserve per-key hashing for anchored intervals or on-demand investigation.

Strategy Latency Coverage Cost Compliance / regulatory
Interval count reconciliation Per interval Net cardinality drift only Low; a windowed count Moderate: proves totals, not per-row fidelity.
Per-key hash reconciliation Per interval / on demand Full value-level divergence High; folds the compacted topic Strongest: a NIST-approved hash gives an auditable per-key proof.
Sampled key reconciliation Continuous Probabilistic value coverage Low; a keyed sample Moderate: bounds risk, cannot prove completeness.
Tombstone audit Per interval Deletes correctly removed Low; scans tombstones Strong: proves regulated deletions actually propagated.

Key Implementation Notes

  • Resolve “latest” by source offset, not consumption order. A partition reassignment can deliver an older record after a newer one; folding by source LSN makes the result independent of consumption order and therefore deterministic.
  • Tombstones prove deletion, not staleness. A missing key must be represented by a tombstone that removes it from the folded state; treating an absent key as “unchanged” hides a failed delete, which is a compliance defect when the deletion was a regulated erasure.
  • Hash a canonical serialization. Reuse the deterministic serializer from column-level checksum generation so a value that is logically equal across engines does not read as divergence because of key ordering or numeric formatting.
  • Anchor the reconciliation interval to a source position. Pin each pass to a source LSN so “the topic matched the source as of position P” is a precise, replayable claim, not a comparison against a moving target.

Verification

Assert that compaction folding resolves by offset, tombstones remove keys, and a value mismatch is caught.

python
records = [
    TopicRecord("k1", {"v": 1}, 10),
    TopicRecord("k1", {"v": 2}, 20),      # newer by offset -> wins
    TopicRecord("k2", {"v": 9}, 15),
    TopicRecord("k2", None, 25),          # tombstone -> k2 removed
]
topic = fold_compacted(records)
assert set(topic) == {"k1"}               # k2 deleted, k1 at latest offset
source = {"k1": canonical_hash({"v": 2}), "k3": canonical_hash({"v": 3})}
report = reconcile(topic, source)
assert report["missing_in_topic"] == ["k3"] and not report["value_mismatch"]
logger.info("kafka topic reconciliation verified")

Operational Considerations

Run count reconciliation continuously and schedule the full hash fold during compaction-quiet windows so a transiently uncompacted topic does not produce phantom mismatches — or read only committed, compacted segments. Fold state is O(distinct keys), so shard by key range across consumers and merge per-shard maps for very large topics, and expose value_mismatch_total, missing_in_topic_total, and tombstone_audit_failures so a regulated deletion that failed to propagate pages immediately. Persist each pass with its anchoring source position and the hash algorithm used, and drive confirmed divergences into the discrepancy routing and remediation path rather than resolving them inline.