build-cdc-streaming-pipeline

name	build-cdc-streaming-pipeline
description	Designs change-data-capture and streaming pipelines — log-based CDC off a DB transaction log (Debezium/WAL/binlog), topic-per-table fan-out onto Kafka/Kinesis, consumer-group/offset/rebalance correctness, windowed/stateful stream processing with watermarks, exactly-once vs at-least-once-plus-idempotent delivery, and Avro/Protobuf schema-registry evolution.
when_to_use	When row changes (incl. deletes) must propagate continuously and low-latency rather than on a schedule — capturing off a transaction log, fanning onto a partitioned stream bus, consuming with correct offset/rebalance/ordering, windowed joins/aggregations, and sinking to a search index/warehouse/cache kept in sync. Distinct from build-etl-pipeline (scheduled batch/incremental loads) and message-queue-jobs (durable server-to-server task queues, not a replayable change log).

When to Use

Reach for this skill when data must flow as a continuous change stream, not land in scheduled batches:

• "Stream every row change out of Postgres/MySQL into Kafka and keep Elasticsearch in sync"
• "Mirror a table into the warehouse in near-real-time, including deletes"
• "My consumer is reprocessing / skipping events after a deploy or rebalance"
• "Consumer-group lag is climbing; ordering is wrong; one partition is hot"
• "Join an orders stream against an enrichment stream with a 5-minute window"
• "Late/out-of-order events are dropped or double-counted"
• "Producer schema changed and consumers broke" / "map Debezium op codes to upserts and deletes"

NOT this skill:

• Scheduled/incremental batch loads to a warehouse (Airflow/dbt, nightly, updated_at cursor) → build-etl-pipeline
• Durable server-to-server work/task queue (enqueue a job, one worker runs it once) → message-queue-jobs
• Client-facing live push over WebSocket/SSE (chat, dashboards) → build-realtime-channel
• Offline client store + delta pull + conflict resolution → build-offline-first-sync
• The replication slot / logical-decoding DDL impact on the source DB itself → db-migration-safety
• Embedding/indexing documents for retrieval as the sink semantics → rag-pipeline

Steps

1. Confirm it's actually streaming, then capture log-based — not query polling. If freshness tolerance is minutes/hours and deletes don't need to propagate, stop and use build-etl-pipeline. If changes (incl. deletes) must land in seconds, do CDC. Pick the capture method:

   Method	Captures deletes	Source load	Ordering	Use when
   Query polling (WHERE updated_at > :cursor)	❌ no (row is gone)	full table scan / index pressure	by updated_at only	no log access; deletes don't matter; small tables
   Log-based CDC (Debezium on WAL/binlog/redo)	✅ yes	low — reads the log the DB already writes	exact commit order per table	default — full fidelity, deletes, low impact
   Trigger-based	✅ yes	write amplification on every DML	by trigger	log unavailable but deletes needed

   Default: Debezium connectors — Postgres (pgoutput logical decoding + replication slot), MySQL (binlog, binlog_format=ROW, binlog_row_image=FULL), Mongo (change streams). Set Postgres wal_level=logical, REPLICA IDENTITY FULL on tables whose before-image (for deletes/diffs) you need.

2. Get the snapshot→stream handoff right, or you lose or double rows at startup. A new connector must read existing rows (snapshot) then switch to live log without a gap. Use snapshot.mode=initial (snapshot once, then stream) — the connector records the log position at snapshot start and streams from there. For huge tables use incremental snapshot (signal-driven, chunked) so streaming isn't blocked and the connector is resumable. Never drop the replication slot while paused — Postgres then discards WAL the connector hasn't read and you get a permanent gap (full re-snapshot required). Monitor pg_replication_slots.confirmed_flush_lsn; an abandoned slot also pins WAL and fills the disk.

3. Shape the bus: topic-per-table, partition key = entity id, choose retention vs compaction. One topic per source table/aggregate (server.table → ). Partition so all events for one entity land on one partition → per-entity ordering is preserved; Kafka guarantees order , never across. Do not key by a low-cardinality column (creates hot partitions) or leave keys null (round-robin → ordering lost).

Common Errors

• enable.auto.commit=true treated as exactly-once. It's a timer that commits independent of your handler — a crash loses or reprocesses. Set it false and commit after the sink flush.
• Committing the offset before the side effect is durable. Crash in the gap = silent data loss. Strict order: process → flush sink → commit.
• Dropping/recreating the Postgres replication slot to "reset". WAL the connector hasn't consumed is discarded → permanent gap, forces a full re-snapshot. Pause the connector, keep the slot; never delete a slot with unconsumed WAL.
• Abandoned/lagging slot fills the source disk. A stopped consumer pins WAL forever. Alert on confirmed_flush_lsn lag and slot age; clean up dead connectors.
• Null or low-cardinality partition key. Null key → round-robin → cross-partition reordering of one entity's events. Low-cardinality key → hot partition. Key by primary key.
• Increasing partition count on a live keyed topic. Rehashes keys → an entity's new events go to a different partition than its in-flight ones → ordering broken. Plan partition count up front; treat increases as a migration.
• Treating a Debezium d (delete) as an upsert. Resurrects deleted rows in the sink. Emit a tombstone (value=null) and let the sink delete; use the ExtractNewRecordState SMT.
• Reading uncommitted transactional records. Without isolation.level=read_committed, consumers see aborted-transaction records and double-count. Set it whenever producers use transactions.
• Poison record retried in place. One un-processable record halts its partition forever and lag explodes. Bounded retries → DLQ topic → commit past it.
• Processing on the poll thread longer than max.poll.interval.ms. Broker thinks the consumer died, rebalances mid-batch, you reprocess. Shrink max.poll.records or raise the interval; offload slow work.
• Eager (stop-the-world) rebalance assignor by default. Every scale event pauses the whole group. Use CooperativeStickyAssignor.
• Windowing on processing time. Replay and out-of-order delivery silently corrupt aggregates. Window on event time with a watermark; route past-grace events to a side-output.
• Raw JSON with no registry. A producer field rename breaks every consumer with no guardrail. Use Avro/Protobuf + registry with BACKWARD compatibility.

Verify

1. Capture fidelity incl. deletes: INSERT, UPDATE, then DELETE a row on the source → consumer observes a create, an update (with correct before/after), and a tombstone, in commit order. A delete that produces no tombstone is a fail.
2. Snapshot→stream no-gap: seed N rows, start the connector, then write M more during the snapshot → exactly N+M distinct rows arrive downstream, none missing, none duplicated past idempotency.
3. Per-entity ordering: rapidly emit 3 updates to one PK → consumer receives them in source order on a single partition (events for that key never interleave out of order).
4. Offset correctness across restart: kill the consumer mid-batch, restart → no committed-but-unprocessed record is lost and no already-sunk record corrupts the sink (idempotency holds). Lag returns to ~0.
5. Rebalance correctness: add then remove a consumer under load with CooperativeStickyAssignor → no record is processed by two members and none is skipped; only moved partitions are revoked (check logs).
6. Replay = same result: --reset-offsets --to-earliest and reprocess → final sink state is byte-identical to before the replay (proves the sink is idempotent/transactional).
7. Poison handling: inject a record the sink rejects → it lands in the DLQ with origin headers, the partition keeps flowing, lag does not climb.
8. Late event: emit an event with an event-time inside a closed-but-within-grace window → the window result updates; past grace → it appears in the side-output, not silently dropped.
9. Schema evolution: register a new schema adding a field with a default under BACKWARD compatibility → old consumers keep running; attempt an incompatible change → registry rejects the register (does not reach consumers).
10. Lag SLO: under sustained source write load, consumer-group lag stays bounded (returns toward 0), not monotonically rising.

Done = deletes propagate as tombstones, snapshot→stream is gap-free, per-entity ordering holds on one partition, a kill-restart and a full replay both leave the sink state correct (idempotent or transactional), poison records go to a DLQ without blocking the partition, late events hit grace/side-output (never silently dropped), and an incompatible schema is rejected at the registry before it reaches any consumer.