CDC Pipelines with Debezium
When to Use
Activate this skill when the task involves:
- Setting up Debezium connectors for PostgreSQL, MySQL, or Oracle
- Interpreting or transforming CDC change events (insert/update/delete/truncate)
- Designing CDC pipelines: database → Kafka → data lake / warehouse
- Implementing the Outbox pattern for transactional CDC
- Troubleshooting replication lag, snapshot failures, or schema evolution
- Integrating CDC streams with Apache Flink, Spark Structured Streaming, or dbt
Core Architecture
┌──────────────────────────────────────────────────────────────┐
│ Source Database │
│ ┌─────────────┐ WAL / binlog / redo log │
│ │ PostgreSQL │──────────────────────┐ │
│ │ MySQL │ ▼ │
│ │ Oracle │ ┌───────────────────────┐ │
│ └─────────────┘ │ Kafka Connect Worker │ │
│ │ ┌─────────────────┐ │ │
│ │ │ Debezium Source │ │ │
│ │ │ Connector │ │ │
│ │ └────────┬────────┘ │ │
│ └───────────│───────────┘ │
│ ▼ │
│ ┌───────────────────────┐ │
│ │ Kafka Topic │ │
│ │ <prefix>.<db>.<tbl> │ │
│ └───────────┬───────────┘ │
│ ▼ │
│ ┌──────────────────────────────┐ │
│ │ Consumers │ │
│ │ • Kafka Connect Sink │ │
│ │ • Apache Flink / Spark │ │
│ │ • ksqlDB / Kafka Streams │ │
│ └──────────────────────────────┘ │
└──────────────────────────────────────────────────────────────┘
Debezium reads database transaction logs (WAL for PostgreSQL, binlog for MySQL), not the tables themselves — zero query load on the source.
Change Event Structure
Every Debezium event is a Kafka message with a structured envelope:
{
"schema": { ... },
"payload": {
"before": {
"id": 1001,
"name": "Alice",
"email": "alice@example.com",
"updated_at": 1700000000000
},
"after": {
"id": 1001,
"name": "Alice Smith",
"email": "alice@example.com",
"updated_at": 1700001000000
},
"source": {
"version": "2.5.0.Final",
"connector": "postgresql",
"name": "pgserver1",
"ts_ms": 1700001000123,
"snapshot": "false",
"db": "mydb",
"sequence": "[\"24023119\",\"24023255\"]",
"schema": "public",
"table": "customers",
"txId": 756,
"lsn": 24023255,
"xmin": null
},
"op": "u",
"ts_ms": 1700001000456,
"transaction": {
"id": "756:24023255",
"total_order": 1,
"data_collection_order": 1
}
}
}
Operation Types
op | Meaning | before | after |
|---|
c | INSERT (create) | null | row state |
u | UPDATE | previous state | new state |
d | DELETE | previous state | null |
r | READ (snapshot) | null | row state |
t | TRUNCATE | null | null |
Tombstone Events
After a d (delete) event, Debezium emits a tombstone — a message with null value and the same key — enabling Kafka log compaction to eventually remove the record.
PostgreSQL Connector
Prerequisites
-- postgresql.conf: must be set before Debezium can connect
ALTER SYSTEM SET wal_level = 'logical';
ALTER SYSTEM SET max_replication_slots = 10;
ALTER SYSTEM SET max_wal_senders = 10;
-- Create dedicated Debezium user
CREATE ROLE debezium WITH LOGIN PASSWORD 'secret' REPLICATION;
GRANT CONNECT ON DATABASE mydb TO debezium;
GRANT USAGE ON SCHEMA public TO debezium;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO debezium;
ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO debezium;
-- Create publication (pgoutput plugin — no extra extension needed)
CREATE PUBLICATION dbz_pub FOR TABLE customers, orders, products;
-- Or capture all tables:
CREATE PUBLICATION dbz_pub FOR ALL TABLES;
Restart PostgreSQL after changing wal_level.
Connector Configuration
{
"name": "pg-customers-connector",
"config": {
"connector.class": "io.debezium.connector.postgresql.PostgresConnector",
"tasks.max": "1",
"database.hostname": "postgres-host",
"database.port": "5432",
"database.user": "debezium",
"database.password": "secret",
"database.dbname": "mydb",
"topic.prefix": "pgserver1",
"plugin.name": "pgoutput",
"publication.name": "dbz_pub",
"slot.name": "debezium_slot",
"table.include.list": "public.customers,public.orders",
"snapshot.mode": "initial",
"snapshot.locking.mode": "minimal",
"heartbeat.interval.ms": "30000",
"heartbeat.topics.prefix": "__debezium-heartbeat",
"key.converter": "io.confluent.kafka.serializers.KafkaAvroSerializer",
"key.converter.schema.registry.url": "http://schema-registry:8081",
"value.converter": "io.confluent.kafka.serializers.KafkaAvroSerializer",
"value.converter.schema.registry.url": "http://schema-registry:8081",
"transforms": "unwrap",
"transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
"transforms.unwrap.drop.tombstones": "false",
"transforms.unwrap.delete.handling.mode": "rewrite",
"transforms.unwrap.add.fields": "op,table,lsn,source.ts_ms",
"transforms.unwrap.add.headers": "db",
"errors.tolerance": "all",
"errors.deadletterqueue.topic.name": "dlq.pgserver1.customers",
"errors.deadletterqueue.topic.replication.factor": "3",
"errors.deadletterqueue.context.headers.enable": "true"
}
}
Snapshot Modes (PostgreSQL)
| Mode | Description | When to Use |
|---|
initial | Snapshot on first start only; then stream | Default — new deployments |
initial_only | Snapshot then stop — no streaming | One-time historical load |
no_data | Skip snapshot, start streaming from now | Append-only use cases |
always | Snapshot on every connector restart | Dev/testing only |
exported | Consistent snapshot without table locks (PG 15+) | Large tables in production |
incremental | Re-snapshot selected tables without stopping | Add new tables mid-stream |
Replication Slot Monitoring
-- Check lag: drop-behind LSN distance
SELECT
slot_name,
pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS lag_size,
active,
database
FROM pg_replication_slots
WHERE slot_type = 'logical';
-- Alert if lag_size > 5GB — Debezium consumer is too slow
MySQL Connector
Prerequisites
# my.cnf — server-id must be unique across all replicas
server-id = 12345
log_bin = mysql-bin
binlog_format = ROW
binlog_row_image = FULL
expire_logs_days = 7
gtid_mode = ON # recommended
enforce_gtid_consistency = ON
CREATE USER 'debezium'@'%' IDENTIFIED BY 'secret';
GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT
ON *.* TO 'debezium'@'%';
FLUSH PRIVILEGES;
Connector Configuration
{
"name": "mysql-orders-connector",
"config": {
"connector.class": "io.debezium.connector.mysql.MySqlConnector",
"tasks.max": "1",
"database.hostname": "mysql-host",
"database.port": "3306",
"database.user": "debezium",
"database.password": "secret",
"database.server.id": "12345",
"topic.prefix": "mysqlserver1",
"database.include.list": "shop",
"table.include.list": "shop.orders,shop.order_items",
"snapshot.mode": "initial",
"snapshot.locking.mode": "minimal",
"include.schema.changes": "true",
"schema.history.internal.kafka.topic": "schema-changes.shop",
"schema.history.internal.kafka.bootstrap.servers": "kafka:9092",
"key.converter": "io.confluent.kafka.serializers.KafkaAvroSerializer",
"key.converter.schema.registry.url": "http://schema-registry:8081",
"value.converter": "io.confluent.kafka.serializers.KafkaAvroSerializer",
"value.converter.schema.registry.url": "http://schema-registry:8081",
"transforms": "unwrap",
"transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
"transforms.unwrap.drop.tombstones": "false",
"transforms.unwrap.delete.handling.mode": "rewrite",
"transforms.unwrap.add.fields": "op,table,source.ts_ms"
}
}
MySQL requires schema.history.internal.kafka.topic — a durable log of all DDL changes. Never delete this topic.
Single Message Transforms (SMTs)
ExtractNewRecordState — Flattening Events
Without SMT, downstream consumers receive the full before/after/source envelope. ExtractNewRecordState flattens it:
Before SMT:
{"payload": {"before": null, "after": {"id": 1, "name": "Alice"}, "op": "c"}}
After SMT:
{"id": 1, "name": "Alice", "__op": "c", "__table": "customers", "__source_ts_ms": 1700001000123}
Field Masking for PII
"transforms": "mask_email",
"transforms.mask_email.type": "org.apache.kafka.connect.transforms.MaskField$Value",
"transforms.mask_email.fields": "email,phone",
"transforms.mask_email.replacement": "***"
Routing to Per-Table Topics
"transforms": "route",
"transforms.route.type": "io.debezium.transforms.ByLogicalTableRouter",
"transforms.route.topic.regex": "pgserver1\\.public\\.(.*)",
"transforms.route.topic.replacement": "cdc.raw.$1"
Kafka Connect Deployment
Docker Compose (Standalone)
version: "3.8"
services:
kafka:
image: confluentinc/cp-kafka:7.6.0
environment:
KAFKA_NODE_ID: 1
KAFKA_PROCESS_ROLES: broker,controller
KAFKA_CONTROLLER_QUORUM_VOTERS: "1@kafka:9093"
KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
CLUSTER_ID: "MkU3OEVBNTcwNTJENDM2Qk"
schema-registry:
image: confluentinc/cp-schema-registry:7.6.0
environment:
SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: kafka:9092
SCHEMA_REGISTRY_HOST_NAME: schema-registry
ports:
- "8081:8081"
kafka-connect:
image: debezium/connect:2.5
environment:
BOOTSTRAP_SERVERS: kafka:9092
GROUP_ID: debezium-connect-cluster
CONFIG_STORAGE_TOPIC: connect-configs
OFFSET_STORAGE_TOPIC: connect-offsets
STATUS_STORAGE_TOPIC: connect-statuses
CONFIG_STORAGE_REPLICATION_FACTOR: 1
OFFSET_STORAGE_REPLICATION_FACTOR: 1
STATUS_STORAGE_REPLICATION_FACTOR: 1
CONNECT_KEY_CONVERTER: io.confluent.kafka.serializers.KafkaAvroSerializer
CONNECT_VALUE_CONVERTER: io.confluent.kafka.serializers.KafkaAvroSerializer
CONNECT_KEY_CONVERTER_SCHEMA_REGISTRY_URL: http://schema-registry:8081
CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL: http://schema-registry:8081
ports:
- "8083:8083"
depends_on:
- kafka
- schema-registry
Connector Lifecycle via REST API
# Deploy connector
curl -s -X POST http://localhost:8083/connectors \
-H "Content-Type: application/json" \
-d @pg-customers-connector.json
# Check status
curl -s http://localhost:8083/connectors/pg-customers-connector/status | jq
# Pause / Resume (drains without losing offset)
curl -X PUT http://localhost:8083/connectors/pg-customers-connector/pause
curl -X PUT http://localhost:8083/connectors/pg-customers-connector/resume
# Restart failed task
curl -X POST "http://localhost:8083/connectors/pg-customers-connector/tasks/0/restart"
# List all connectors
curl -s http://localhost:8083/connectors | jq
# Delete connector (releases replication slot!)
curl -X DELETE http://localhost:8083/connectors/pg-customers-connector
Outbox Pattern
The Outbox pattern guarantees exactly-once delivery of domain events without dual writes or distributed transactions.
PostgreSQL Setup
-- Outbox table lives in the same database as the domain table
CREATE TABLE outbox (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
aggregate_type VARCHAR(255) NOT NULL, -- e.g. 'Order'
aggregate_id VARCHAR(255) NOT NULL, -- e.g. order_id
event_type VARCHAR(255) NOT NULL, -- e.g. 'OrderPlaced'
payload JSONB NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
-- Application writes domain change + outbox row in ONE transaction
BEGIN;
INSERT INTO orders (id, customer_id, total) VALUES (42, 7, 199.99);
INSERT INTO outbox (aggregate_type, aggregate_id, event_type, payload)
VALUES ('Order', '42', 'OrderPlaced',
'{"order_id": 42, "customer_id": 7, "total": 199.99}');
COMMIT;
Outbox Connector Configuration
{
"name": "outbox-connector",
"config": {
"connector.class": "io.debezium.connector.postgresql.PostgresConnector",
"topic.prefix": "outbox",
"table.include.list": "public.outbox",
"plugin.name": "pgoutput",
"publication.name": "dbz_outbox_pub",
"slot.name": "debezium_outbox_slot",
"transforms": "outbox",
"transforms.outbox.type": "io.debezium.transforms.outbox.EventRouter",
"transforms.outbox.table.field.event.id": "id",
"transforms.outbox.table.field.event.key": "aggregate_id",
"transforms.outbox.table.field.event.type": "event_type",
"transforms.outbox.table.field.event.payload": "payload",
"transforms.outbox.route.by.field": "aggregate_type",
"transforms.outbox.route.topic.replacement": "events.${routedByValue}.Avro",
"transforms.outbox.table.expand.json.payload": "true",
"tombstones.on.delete": "false"
}
}
The EventRouter SMT routes each row to events.Order.Avro, events.Payment.Avro, etc. and tombstones deleted outbox rows (after cleanup jobs purge them).
Consuming CDC Events — Python
from confluent_kafka import Consumer, KafkaException
from confluent_kafka.schema_registry.avro import AvroDeserializer
from confluent_kafka.serialization import SerializationContext, MessageField
import json
consumer = Consumer({
"bootstrap.servers": "kafka:9092",
"group.id": "cdc-consumer-group",
"auto.offset.reset": "earliest",
"enable.auto.commit": False,
"max.poll.interval.ms": 300000,
})
consumer.subscribe(["pgserver1.public.customers"])
try:
while True:
msg = consumer.poll(timeout=1.0)
if msg is None:
continue
if msg.error():
raise KafkaException(msg.error())
# With ExtractNewRecordState SMT, value is a flat dict
value = json.loads(msg.value())
op = value.get("__op", "r")
if op == "c":
upsert_to_sink(value)
elif op == "u":
upsert_to_sink(value)
elif op == "d":
delete_from_sink(value["id"])
consumer.commit(asynchronous=False)
finally:
consumer.close()
Apache Flink CDC Integration
Flink SQL can consume Debezium-formatted Kafka topics natively via the debezium-json format.
-- Flink SQL: source table reading Debezium JSON from Kafka
CREATE TABLE customers_cdc (
id BIGINT,
name STRING,
email STRING,
updated_at TIMESTAMP(3),
PRIMARY KEY (id) NOT ENFORCED
) WITH (
'connector' = 'kafka',
'topic' = 'pgserver1.public.customers',
'properties.bootstrap.servers' = 'kafka:9092',
'properties.group.id' = 'flink-cdc-group',
'scan.startup.mode' = 'earliest-offset',
'format' = 'debezium-json',
'debezium-json.schema-include' = 'false'
);
-- Flink SQL: Iceberg sink (upsert semantics)
CREATE TABLE customers_iceberg (
id BIGINT,
name STRING,
email STRING,
PRIMARY KEY (id) NOT ENFORCED
) WITH (
'connector' = 'iceberg',
'catalog-name' = 'rest',
'catalog-type' = 'rest',
'uri' = 'http://iceberg-rest:8181',
'warehouse' = 's3://lakehouse/warehouse',
'database-name' = 'silver',
'table-name' = 'customers'
);
INSERT INTO customers_iceberg
SELECT id, name, email FROM customers_cdc;
For Avro with Schema Registry, use 'format' = 'debezium-avro-confluent' and set 'debezium-avro-confluent.url'.
Iceberg / Delta Sink via Kafka Connect
Iceberg Sink Connector (Tabular)
{
"name": "iceberg-sink-customers",
"config": {
"connector.class": "io.tabular.iceberg.connect.IcebergSinkConnector",
"tasks.max": "2",
"topics": "pgserver1.public.customers",
"iceberg.catalog.type": "rest",
"iceberg.catalog.uri": "http://iceberg-rest:8181",
"iceberg.catalog.warehouse": "s3://lakehouse/warehouse",
"iceberg.tables": "silver.customers",
"iceberg.tables.upsert-mode-enabled": "true",
"iceberg.tables.cdc-field": "__op",
"iceberg.control.topic": "iceberg-control",
"iceberg.control.group-id": "iceberg-sink-group",
"iceberg.commit.interval-ms": "60000",
"iceberg.commit.timeout-ms": "30000"
}
}
Schema Evolution Handling
Debezium propagates DDL changes automatically. Key behaviors:
| Change | PostgreSQL (pgoutput) | MySQL |
|---|
ADD COLUMN | New field appears in after | Schema history updated; new events include column |
DROP COLUMN | Field disappears from after | Schema history updated |
RENAME COLUMN | Treat as drop + add (no rename event) | Same |
ALTER COLUMN TYPE | Supported for compatible types | Supported |
TRUNCATE TABLE | op: "t" event emitted | Emits op: "t" |
For downstream Iceberg/Delta sinks, enable schema evolution on the sink connector to auto-add new columns rather than failing.
Monitoring
Key JMX Metrics
| Metric | Connector Type | Alert Threshold |
|---|
MilliSecondsBehindSource | All | > 60,000 ms |
NumberOfCommittedTransactions | MySQL | Rate drop to 0 |
SnapshotRunning | All | Stuck > expected time |
TotalNumberOfErrorsSeen | All | > 0 |
QueueRemainingCapacity | All | < 10% |
PostgreSQL Replication Lag
SELECT
application_name,
pg_size_pretty(pg_wal_lsn_diff(
sent_lsn, replay_lsn
)) AS replay_lag,
state
FROM pg_stat_replication;
Heartbeat Configuration
Without heartbeats, a quiet source database (no writes) will not advance the replication slot LSN, causing WAL accumulation.
"heartbeat.interval.ms": "30000",
"heartbeat.action.query": "INSERT INTO debezium_heartbeat VALUES (DEFAULT) ON CONFLICT DO NOTHING"
Create the heartbeat table:
CREATE TABLE IF NOT EXISTS debezium_heartbeat (
id SERIAL PRIMARY KEY,
ts TIMESTAMPTZ DEFAULT now()
);
Anti-Patterns
-
Not monitoring replication slot lag — WAL accumulates unboundedly; PostgreSQL disk fills up. Set alert at 5 GB lag.
-
Deleting the replication slot externally — kills the connector offset; forces full re-snapshot. Never drop slots manually.
-
Using snapshot.mode: always in production — re-snapshots entire tables on every restart, causing massive traffic and duplicates downstream.
-
One connector for all tables — a single failed task blocks all tables. Split high-throughput or high-risk tables into separate connectors.
-
Missing heartbeat on write-quiet databases — WAL accumulates when source tables have no writes. Always configure heartbeat.interval.ms.
-
Not enabling the DLQ — deserialization or routing errors silently stop the connector task. Always configure errors.tolerance: all + errors.deadletterqueue.topic.name.
-
Dual writes (app writes DB + Kafka) — violates atomicity. Use the Outbox pattern instead.
-
Using schema_only snapshot mode (deprecated) — replaced by no_data. Using it produces a warning and may not be supported in future versions.
-
No tombstones.on.delete handling in sink — sinks that don't process tombstones accumulate ghost records in compacted topics.
-
Transforming events inside Kafka Connect for complex logic — SMTs are for simple field mapping only. Route raw events to Kafka and apply transformation in Flink or Spark.
References to Consult When Needed
- Debezium PostgreSQL connector reference:
debezium.io/documentation/reference/stable/connectors/postgresql.html
- Debezium MySQL connector reference:
debezium.io/documentation/reference/stable/connectors/mysql.html
- ExtractNewRecordState SMT:
debezium.io/documentation/reference/stable/transformations/event-flattening.html
- Outbox EventRouter SMT:
debezium.io/documentation/reference/stable/transformations/outbox-event-router.html
- Flink Debezium format:
nightlies.apache.org/flink/flink-docs-stable/docs/connectors/table/formats/debezium/