Streaming POS Data with Change Data Capture
Batch extracts of point-of-sale data arrive hours late, miss same-day voids and price corrections, and re-import rows the reconciliation engine has already accrued against — every one of those gaps is a rebate variance waiting to surface at month-end. Log-based change data capture (CDC) closes the gap by tailing the POS database’s write-ahead log and emitting every insert, update, and delete as an ordered event the moment it commits, so the claim validation rule engine evaluates transactions in near-real time instead of against a stale snapshot. This page is the implementation-level companion to the streaming and CDC ingestion topic area, which sits inside data ingestion normalization pipelines; here we build one concrete flow — a Debezium-style connector emitting to a partitioned event-log topic, consumed idempotently so replays and duplicate deliveries change nothing.
Prerequisites
Before wiring CDC into the reconciliation pipeline, confirm the following are in place:
- A log-based CDC connector with source access. A Debezium-style connector needs replication access to the POS database — logical replication on PostgreSQL (
REPLICATIONrole plus a replication slot), orREPLICATION CLIENT/REPLICATION SLAVEand row-based binlog on MySQL. Trigger-based or query-based capture is out of scope; only the write-ahead log gives you complete, ordered before/after images including deletes. - A partitioned event-log topic and an offset store. Provision the Kafka topic
pos.public.transactionspartitioned so that a transaction’s whole lifecycle lands on one partition, and a durable consumer-offset store (the broker’s__consumer_offsetsor an external table) that survives restarts. Per-key ordering and replay both depend on this. - A schema registry holding the change-event contract. The
op,before, andafterfield types are registered once so the connector and every consumer serialize and validate against one definition instead of drifting per service. - Python 3.11+ with
pydantic>=2.5. All validation snippets use Pydantic v2 models. Every monetary field is parsed viadecimal.Decimal; neverfloat. - Scoped roles. The connector runs under a read-only replication role on the source; the consumer runs under a
reconciliation_etlrole with write access to the materialized transaction table and its own offsets, and read-only access to the schema registry. Neither role can post accruals — that boundary belongs downstream in settlement and financial close.
Step-by-Step Implementation
Step 1 — Model the change event
A CDC message is not a row; it is a change — an operation code plus the row’s before and after images and the source coordinates that place it in the log. Encode that contract as a typed Pydantic v2 ChangeEvent so a malformed event fails at deserialization instead of corrupting the materialized table. Amounts are parsed from strings into decimal.Decimal, and a validator rejects any float outright.
from decimal import Decimal
from enum import Enum
from typing import Optional
from datetime import datetime
from pydantic import BaseModel, field_validator, model_validator
class Op(str, Enum):
c = "c" # create / insert
u = "u" # update
d = "d" # delete
class PosRow(BaseModel):
txn_id: str
store_id: str
register_id: str
sku: str
units: int
net_amount: Decimal
txn_ts: datetime
@field_validator("net_amount", mode="before")
@classmethod
def _no_float(cls, v):
if isinstance(v, float):
raise ValueError("net_amount must be a decimal string, not a float")
return v
class Source(BaseModel):
topic: str
partition: int
offset: int
lsn: int # log sequence number from the source WAL
class ChangeEvent(BaseModel):
op: Op
before: Optional[PosRow] = None
after: Optional[PosRow] = None
source: Source
ingest_ts: datetime # when the connector emitted the event
@model_validator(mode="after")
def _shape(self) -> "ChangeEvent":
if self.op in (Op.c, Op.u) and self.after is None:
raise ValueError("create/update events require an 'after' image")
if self.op is Op.d and self.before is None:
raise ValueError("delete events require a 'before' image")
return selfA representative update event on the wire — a cashier corrected the unit count on an already-captured line:
{
"op": "u",
"before": {
"txn_id": "T-99118", "store_id": "S-4471", "register_id": "R-08",
"sku": "0004900000634", "units": 6, "net_amount": "23.94",
"txn_ts": "2026-07-15T18:42:11Z"
},
"after": {
"txn_id": "T-99118", "store_id": "S-4471", "register_id": "R-08",
"sku": "0004900000634", "units": 4, "net_amount": "15.96",
"txn_ts": "2026-07-15T18:42:11Z"
},
"source": {"topic": "pos.public.transactions", "partition": 3, "offset": 88214, "lsn": 27713994},
"ingest_ts": "2026-07-15T18:42:12.104Z"
}Validation check: parse the payload with ChangeEvent.model_validate_json(raw) and assert event.after.net_amount == Decimal("15.96"). Then feed the same payload with "net_amount": 15.96 (a JSON number) and assert it raises ValidationError. A float that reaches the accrual table is a rounding defect you cannot reproduce later.
Step 2 — Consume in offset order per key
Correctness hinges on ordering: for one transaction, its create must be applied before its update, and its update before its delete. A log-based connector guarantees this only within a partition, so the routing key must place every event for a transaction on the same partition. Derive a deterministic key and process events strictly in ascending offset, tracking the last applied offset per partition so an out-of-order or already-seen event is dropped rather than re-applied.
def key_of(evt: ChangeEvent) -> str:
row = evt.after or evt.before # 'after' for c/u, 'before' for d
return f"{row.store_id}:{row.register_id}:{row.txn_id}"
class OffsetGuard:
def __init__(self):
self._applied: dict[int, int] = {} # partition -> last applied offset
def accept(self, evt: ChangeEvent) -> bool:
part, off = evt.source.partition, evt.source.offset
if off <= self._applied.get(part, -1):
return False # stale / duplicate by offset
self._applied[part] = off
return TrueBecause the key hashes to a single partition and the broker preserves per-partition offset order, iterating a partition in offset order replays the transaction’s lifecycle in commit order. The guard makes redelivery of an already-processed offset a no-op.
Validation check: submit events for the same key_of(...) at offsets 10, 11, then a redelivered 11, and assert accept returns True, True, False. Then submit offset 9 after 11 and assert it returns False — a late, out-of-order frame must not overwrite newer state.
Step 3 — Idempotent upsert keyed on a deterministic record_hash
Offset guarding stops same-partition redelivery, but a full topic replay resets offsets and re-delivers everything. The durable defense is content-addressed idempotency: fingerprint each row into a deterministic record_hash and make the write a no-op when the incoming hash equals the stored one. Serialize with sorted keys and the amount as a string so the hash is stable across processes and machines.
import hashlib
import json
def record_hash(row: PosRow) -> str:
payload = {
"txn_id": row.txn_id,
"store_id": row.store_id,
"register_id": row.register_id,
"sku": row.sku,
"units": row.units,
"net_amount": str(row.net_amount), # decimal string, never float repr
"txn_ts": row.txn_ts.isoformat(),
}
blob = json.dumps(payload, sort_keys=True, separators=(",", ":"))
return hashlib.sha256(blob.encode()).hexdigest()
class IdempotentSink:
def __init__(self):
self._rows: dict[str, dict] = {} # txn key -> {record_hash, row}
def upsert(self, evt: ChangeEvent) -> str:
row = evt.after
key, rhash = key_of(evt), record_hash(row)
current = self._rows.get(key)
if current and current["record_hash"] == rhash:
return "noop_duplicate" # replay of identical content
self._rows[key] = {"record_hash": rhash, "row": row}
return "inserted" if current is None else "updated"In a relational sink this is one statement: INSERT ... ON CONFLICT (txn_key) DO UPDATE SET ... WHERE pos.record_hash <> excluded.record_hash. A replayed create writes identical content, the hashes match, and the row is untouched — the accrual downstream never double-counts.
Validation check: apply a create, then apply the same create again and assert the second call returns "noop_duplicate". Apply the Step 1 update event and assert it returns "updated" and that record_hash changed with units. Idempotency means N deliveries of one event leave exactly one materialized row.
Step 4 — Handle tombstone deletes
A voided or reversed sale must remove the row, or the reconciliation engine accrues rebate against a transaction that no longer exists. A log-based connector represents a delete as an op: "d" event carrying the before image, immediately followed by a Kafka tombstone — a message with the same key and a null value that lets log compaction physically retire the key. The consumer must act on the delete and treat the tombstone as an idempotent confirmation, not an error.
def apply(sink: IdempotentSink, guard: OffsetGuard, evt: ChangeEvent | None, key: str) -> str:
if evt is None: # raw Kafka tombstone: key set, value null
sink._rows.pop(key, None)
return "tombstone_compacted"
if not guard.accept(evt):
return "skipped_stale_offset"
if evt.op is Op.d:
sink._rows.pop(key_of(evt), None) # remove using the 'before' image key
return "deleted"
return sink.upsert(evt) # c / u -> idempotent upsertDeleting an absent key with pop(key, None) is deliberately safe: whether the delete arrives once or the tombstone re-arrives after a replay, the end state is the same — no row. That is what keeps deletes idempotent alongside upserts.
Validation check: insert a row, apply its op: "d" event, and assert key_of is gone from the sink. Then deliver the tombstone (evt is None) for the same key and assert it returns "tombstone_compacted" without raising. A missed tombstone is the classic cause of phantom accruals on reversed sales.
Step 5 — Apply a watermark for late events and support replay
Events can arrive out of wall-clock order across partitions — a register that buffered offline flushes an hour late. A watermark tracks the high boundary of event time seen so far; anything older than the watermark minus a bounded lag is late and is routed to a correction path instead of silently mutating a period that reconciliation may already be closing. Replay is the complement: because Steps 3 and 4 are idempotent, you can seek a partition back to any committed offset and re-consume with no double-counting.
from datetime import timedelta
WATERMARK_LAG = timedelta(minutes=30) # tolerated out-of-orderness
class Watermark:
def __init__(self):
self._hi: datetime | None = None
def observe(self, evt: ChangeEvent) -> None:
ts = (evt.after or evt.before).txn_ts
if self._hi is None or ts > self._hi:
self._hi = ts
def is_late(self, evt: ChangeEvent) -> bool:
if self._hi is None:
return False
return (evt.after or evt.before).txn_ts < self._hi - WATERMARK_LAG
# Replay: reset a partition to a stored offset; idempotency makes it safe.
def replay_from(consumer, topic: str, partition: int, offset: int) -> None:
from kafka import TopicPartition # kafka-python style API
tp = TopicPartition(topic, partition)
consumer.seek(tp, offset)Late events are not discarded — they carry an accrual delta against the original period, which is exactly the retroactive-adjustment behavior the date-window alignment checks layer expects. The watermark only decides which path the event takes.
Validation check: observe an event at txn_ts = 19:00, then submit one at 18:20 and assert is_late is True (older than 30 minutes behind the watermark) while one at 18:45 is False. After a replay_from(...), re-consume the range and assert the materialized row count is unchanged — replay is a no-op on already-applied content.
Common Failure Modes and Fixes
1. Out-of-order application within a key. Round-robin partitioning, or keying on sku instead of the transaction identity, scatters one transaction’s create/update/delete across partitions where offset order no longer holds — the update can land before the create. Fix: key strictly on the transaction identity (store_id:register_id:txn_id) so the whole lifecycle shares one partition, and enforce the per-partition OffsetGuard from Step 2.
2. Duplicate on redelivery or rebalance. At-least-once delivery re-sends the last uncommitted batch after a consumer rebalance or crash, double-applying events keyed only on offset. Fix: content-address every write with record_hash (Step 3) so an identical redelivery is a noop_duplicate; commit offsets only after the sink write succeeds.
3. Float amounts corrupting the hash and the accrual. Deserializing net_amount as a JSON number reintroduces double-precision drift and makes record_hash non-reproducible, because str(15.96) can differ from the source string. Fix: parse amounts as decimal.Decimal from strings, reject floats at the boundary (Step 1), and serialize with str(row.net_amount) in the hash payload.
4. Missed tombstone leaving a phantom row. Treating an op: "d" event or a null-value tombstone as malformed, or filtering it out because it has no after image, leaves the reversed sale materialized and accruing. Fix: handle op: "d" on the before key and accept the null-value tombstone as an idempotent delete (Step 4).
5. Offset gap or growing consumer lag. A dropped replication slot, an expired offset, or a slow consumer creates a gap where committed data was never applied, or an ever-growing lag that makes reconciliation stale. Fix: monitor consumer lag per partition, alert when it exceeds an SLA, and recover by replay_from the last known-good committed offset — safe precisely because Steps 3 and 4 are idempotent.
Operational Checklist
Frequently Asked Questions
Why use log-based CDC instead of a periodic batch extract of POS data? A batch extract captures a snapshot and loses intermediate states — a line that was created, corrected, and voided between two extracts may never appear, or appears in a final state that hides the correction the vendor was rebated against. Log-based CDC emits every insert, update, and delete in commit order the moment it happens, so voids and price corrections reconcile in near-real time. The trade-off in extract cadence versus completeness is covered in incremental versus full ERP sync strategies.
How does keying on record_hash make replays safe? The record_hash is a stable SHA-256 fingerprint of the row’s content, computed from sorted keys with the amount as a decimal string. An idempotent upsert compares the incoming hash against the stored one and skips the write when they match, so re-consuming a topic from an earlier offset re-applies identical content as a no-op. N deliveries of the same event always leave exactly one materialized row.
What is a tombstone and why does it matter for reconciliation? A tombstone is a Kafka message with a key and a null value that a log-based connector emits after a delete so log compaction can physically retire the key. If the consumer ignores it, the deleted transaction stays materialized and the engine keeps accruing rebate against a sale that was reversed. Handling both the op: "d" event and the null-value tombstone as idempotent deletes prevents that phantom accrual.
How do late-arriving events avoid corrupting a closed period? A watermark tracks the highest event time seen and flags anything older than a bounded lag as late. Late events are not applied blindly into the current window; they are routed to a correction path that posts a delta against the original period, keeping period-over-period accrual tie-out intact rather than silently mutating a period reconciliation may already be closing.
Related
- Parent topic: Streaming & CDC Ingestion — the broader change-data-capture ingestion architecture this flow implements.
- Incremental vs Full ERP Sync Strategies — when a bounded batch sync fits better than a continuous CDC stream, and how the two coexist.
- Implementing Async Batch Queues for Sales Data — buffering and backpressure patterns for the downstream consumers this stream feeds.
Up one level: Streaming & CDC Ingestion