CDC with Debezium skill for Codex

Done well means: every insert, update and delete in the selected tables reaches consumers in per-key order, the source database is never put at risk by a stuck replication slot, snapshots and re-snapshots are planned, and consumers apply events idempotently.

Before configuring

1. Read AGENTS.md, existing connector configs (connectors/*.json, Helm values, Strimzi KafkaConnector resources) and the Kafka Connect worker config.
2. Confirm versions: Debezium (3.x needs Java 17+), the database version, and whether the database is managed (RDS, Cloud SQL, Azure) which changes how you enable logical replication.
3. Decide the runtime:

   Runtime	Use when
   Kafka Connect + Debezium connector	Kafka is the backbone; most common
   Debezium Server	Sink is Kinesis, Pub/Sub, Pulsar, Redis, HTTP without running Kafka
   Embedded engine (Java)	CDC inside one application; you own offsets and restarts

4. Changing database parameters, creating replication users, slots or publications on a production database requires the DBA’s or user’s explicit approval. Some parameters need a restart.

PostgreSQL prerequisites

-- postgresql.conf (or managed-service parameter group): wal_level = logical,
-- max_replication_slots and max_wal_senders above current use, max_slot_wal_keep_size set (PG 13+)
create role debezium with login replication password '...';
grant select on public.orders, public.customers to debezium;
create publication dbz_shop for table public.orders, public.customers;
alter table public.orders replica identity full;   -- only where consumers need full before-images

With the default replica identity, update and delete events carry only primary key columns in before, and unchanged TOASTed columns arrive as a placeholder (__debezium_unavailable_value). Use REPLICA IDENTITY FULL on tables where that matters, knowing it increases WAL volume.

Connector config (PostgreSQL)

{
  "name": "shop-pg",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "pg.internal",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "${file:/opt/secrets/pg.properties:password}",
    "database.dbname": "shop",
    "topic.prefix": "shop",
    "plugin.name": "pgoutput",
    "slot.name": "debezium_shop",
    "publication.name": "dbz_shop",
    "publication.autocreate.mode": "disabled",
    "table.include.list": "public.orders,public.customers",
    "snapshot.mode": "initial",
    "heartbeat.interval.ms": "10000",
    "signal.data.collection": "public.debezium_signal",
    "decimal.handling.mode": "string",
    "tombstones.on.delete": "true"
  }
}

• Topics are named <topic.prefix>.<schema>.<table>, keyed by primary key, so per-row order is preserved within a partition. Tables without a primary key need message.key.columns.
• Secrets via a config provider (config.providers=file on the worker); never inline passwords in JSON committed to git.
• topic.prefix and slot.name are identity: changing them later means a new slot and a full new snapshot.

MySQL differences: binlog_format=ROW, binlog_row_image=FULL, GTIDs on, binlog retention longer than any expected connector downtime, a unique database.server.id, and schema.history.internal.kafka.topic (never delete or compact-away that topic).

Snapshots

snapshot.mode	Behaviour	Use
initial (default)	Snapshot existing rows, then stream	New pipelines
no_data	Stream only from now on (formerly never/schema_only)	History loaded another way
when_needed	Snapshot if offsets are missing or no longer valid	Self-healing after long outages
initial_only	Snapshot then stop	One-off copies

Re-snapshot one table without restarting, using an incremental snapshot signal:

create table public.debezium_signal (id varchar(42) primary key, type varchar(32) not null, data varchar(2048));
insert into public.debezium_signal (id, type, data) values
  ('resnap-orders-2025-09-30', 'execute-snapshot',
   '{"data-collections": ["public.orders"], "type": "incremental"}');

Incremental snapshots read in chunks interleaved with streaming, so they do not block writes. The signal table must be in the publication / include list.

Event shape and flattening

Raw events carry before, after, source (LSN, txId, ts_ms), op (c create, u update, d delete, r snapshot read) and ts_ms. Most sinks want the flat row:

"transforms": "unwrap",
"transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
"transforms.unwrap.delete.tombstone.handling.mode": "rewrite",
"transforms.unwrap.add.fields": "op,source.ts_ms,source.lsn"

rewrite keeps deletes as rows with __deleted=true, which warehouse sinks can apply. Dropping deletes silently is the most common CDC correctness bug.

Transactional outbox

Publish domain events reliably by writing them to an outbox table in the same transaction as the business change, and let Debezium route them.

create table outbox (
  id uuid primary key,
  aggregatetype varchar(255) not null,   -- routes to a topic, e.g. "order"
  aggregateid varchar(255) not null,     -- becomes the message key
  type varchar(255) not null,            -- event type, e.g. "OrderPlaced"
  payload jsonb not null,
  created_at timestamptz not null default now()
);

"table.include.list": "public.outbox",
"transforms": "outbox",
"transforms.outbox.type": "io.debezium.transforms.outbox.EventRouter",
"transforms.outbox.route.topic.replacement": "${routedByValue}.events",
"transforms.outbox.table.expand.json.payload": "true"

Messages are keyed by aggregateid and carry the outbox id as a header; consumers dedupe on it. Purge old outbox rows on a schedule (or delete in the same transaction right after insert; the WAL still carries the insert). For design questions beyond Debezium wiring, see $outbox-pattern.

Operating safely

The main risk is an inactive replication slot retaining WAL until the database disk fills.

select slot_name, active,
       pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn)) as retained
from pg_replication_slots;

• Alert when retained WAL exceeds a threshold (for example 10 GB or 20% of free disk).
• Set max_slot_wal_keep_size so a dead connector invalidates its slot instead of taking down the primary (the connector then needs a re-snapshot, which when_needed can do).
• Low-traffic tables in a busy database: heartbeats plus heartbeat.action.query writing to a heartbeat table keep the confirmed LSN moving.
• Dropping a slot (pg_drop_replication_slot) loses the stream position; get explicit confirmation and plan a re-snapshot first.

Kafka Connect REST checks (worker network access needed):

curl -s http://connect:8083/connectors/shop-pg/status | jq
curl -s -X POST "http://connect:8083/connectors/shop-pg/restart?includeTasks=true&onlyFailed=true"

Downstream apply

• Upsert by primary key and apply deletes; order by source.lsn (Postgres) or binlog position when batching, because one batch can contain several versions of a row.
• Expect duplicates after connector restarts (at-least-once). Idempotent upserts make them harmless.
• Schema changes: Postgres pgoutput carries new columns in events after the change; add them downstream deliberately. Dropping a column upstream should fail loudly in the sink.

Verification and report

• Insert, update and delete a test row in a non-production database; confirm three events with correct op, key and before/after, and that the sink reflects the delete.
• Compare row counts and a checksum of a column between source and sink after the snapshot.
• Watch retained WAL and connector lag for a full business day.

**Connector:** shop-pg (Debezium 3.x, pgoutput, slot debezium_shop, publication dbz_shop)
**Tables:** public.orders (replica identity full), public.customers
**Snapshot:** initial, 41.2M rows in 38 min; counts match source
**Deletes:** rewritten with __deleted=true, applied by sink MERGE
**Safety:** max_slot_wal_keep_size 50GB; alert at 10 GB retained; heartbeat 10 s
**Test:** c/u/d round-trip verified in staging