CDC with Debezium and Kafka: A Source-to-Sink Walkthrough

Change Data Capture (CDC) moves database changes as events instead of repeatedly querying whole tables. In this walkthrough, PostgreSQL is the source of truth, Debezium reads its write-ahead log (WAL), Kafka transports the change events, and a small Python consumer maintains a SQLite projection as the sink.

The example uses a customers table, but the same pattern applies to operational databases feeding search indexes, warehouses, caches, downstream services, or lakehouse ingestion pipelines.

The Architecture and Delivery Model

The pipeline has four responsibilities:

PostgreSQL │ │ logical decoding from WAL ▼ Debezium PostgreSQL connector │ │ Kafka Connect source task ▼ Kafka topic: shop.public.customers │ │ ordered per Kafka partition ▼ Python sink consumer │ ▼ SQLite customer_projection

PostgreSQL records every committed change in its WAL. Debezium does not poll SELECT * FROM customers; instead, it creates or uses a logical replication slot and converts committed WAL changes into Kafka records.

A typical lifecycle is:

Debezium performs an initial snapshot of the configured table.
It records the WAL position associated with that snapshot.
It streams later inserts, updates, and deletes from the WAL.
A sink consumes Kafka records and updates its own representation of the data.

This architecture is usually at-least-once end to end. A consumer can process the same event again after a restart, rebalance, or failure between writing to the sink and committing its Kafka offset. Therefore, sink writes must be idempotent.

For a table-backed sink, the usual idempotent pattern is:

use the source primary key as the Kafka message key;
upsert inserts and updates;
delete by primary key;
commit the Kafka offset only after the sink transaction commits.

Prepare PostgreSQL for Logical Replication

Assume the following:

PostgreSQL is reachable at localhost:5432;
Kafka is reachable at localhost:9092;
Kafka Connect is running at http://localhost:8083;
the Kafka Connect worker has the Debezium PostgreSQL connector plugin installed;
PostgreSQL has wal_level=logical and has been restarted after that setting was changed.

Run the following as a PostgreSQL administrator. The ALTER SYSTEM command requires a database restart before it takes effect.

ALTER SYSTEM SET wal_level = 'logical'; CREATE ROLE debezium WITH LOGIN REPLICATION PASSWORD 'dbz_password'; CREATE DATABASE appdb;

Connect to appdb, then create a source table and initial rows:

CREATE TABLE public.customers ( id INTEGER PRIMARY KEY, full_name TEXT NOT NULL, email TEXT NOT NULL UNIQUE, updated_at TIMESTAMPTZ NOT NULL DEFAULT now() ); INSERT INTO public.customers (id, full_name, email) VALUES (1, 'Aisha Rahman', 'aisha@example.com'), (2, 'Mark Chen', 'mark@example.com'); GRANT USAGE ON SCHEMA public TO debezium; GRANT SELECT ON ALL TABLES IN SCHEMA public TO debezium; CREATE PUBLICATION dbz_publication FOR TABLE public.customers;

The publication tells PostgreSQL which table changes may be emitted through the pgoutput logical decoding plugin.

The primary key is important. Debezium uses it to construct a stable Kafka key such as:

{"id": 1}

Without a stable key, a downstream sink cannot reliably identify which row should be updated or deleted.

Register the Debezium Connector

Kafka Connect runs connectors. Debezium is a source connector: it reads from PostgreSQL and writes to Kafka.

Save the following as customers-connector.json:

{ "name": "customers-cdc", "config": { "connector.class": "io.debezium.connector.postgresql.PostgresConnector", "plugin.name": "pgoutput", "database.hostname": "localhost", "database.port": "5432", "database.user": "debezium", "database.password": "dbz_password", "database.dbname": "appdb", "topic.prefix": "shop", "table.include.list": "public.customers", "publication.name": "dbz_publication", "publication.autocreate.mode": "disabled", "slot.name": "customers_cdc_slot", "snapshot.mode": "initial", "schema.history.internal.kafka.bootstrap.servers": "localhost:9092", "schema.history.internal.kafka.topic": "dbz-schema-history.shop", "key.converter": "org.apache.kafka.connect.json.JsonConverter", "key.converter.schemas.enable": "false", "value.converter": "org.apache.kafka.connect.json.JsonConverter", "value.converter.schemas.enable": "false", "transforms": "unwrap", "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState", "transforms.unwrap.add.fields": "op,source.ts_ms", "transforms.unwrap.delete.handling.mode": "rewrite", "transforms.unwrap.delete.tombstone.handling.mode": "drop" } }

curl -X POST \ -H "Content-Type: application/json" \ --data @customers-connector.json \ http://localhost:8083/connectors

Check connector and task status:

curl http://localhost:8083/connectors/customers-cdc/status

The source topic is:

shop.public.customers

Without the ExtractNewRecordState transformation, Debezium emits a detailed envelope containing before, after, source metadata, transaction context, and operation type. That full envelope is valuable for audit pipelines.

This tutorial unwraps the after state because the sink is maintaining a current-state table. It adds metadata fields such as __op, where values commonly include:

r: snapshot read
c: create
u: update
d: delete

For a broader structured learning path after building this example, this data engineering roadmap is an optional next step.

Generate Changes and Inspect the Kafka Records

Make a few source-side changes:

UPDATE public.customers SET email = 'aisha.rahman@example.com', updated_at = now() WHERE id = 1; INSERT INTO public.customers (id, full_name, email) VALUES (3, 'Omar Haddad', 'omar@example.com'); DELETE FROM public.customers WHERE id = 2;

After unwrapping, the update event is approximately:

{ "id": 1, "full_name": "Aisha Rahman", "email": "aisha.rahman@example.com", "updated_at": "2025-...", "__op": "u", "__source_ts_ms": 1730000000000 }

The exact timestamps will differ. A delete is rewritten into a value containing a deletion marker, typically __deleted: "true", while retaining the Kafka key.

You can inspect records using Kafka CLI tools. The exact command depends on the Kafka distribution, but the essential requirement is to display both key and value:

kafka-console-consumer \ --bootstrap-server localhost:9092 \ --topic shop.public.customers \ --from-beginning \ --property print.key=true \ --property key.separator=" | "

Build a Small Idempotent Python Sink

This consumer writes the topic into SQLite. It is intentionally small, but it demonstrates the core production rule: write to the sink first, then commit the Kafka offset.

Install the dependency:

pip install kafka-python

Save this as customer_sink.py:

import json import sqlite3 from kafka import KafkaConsumer TOPIC = "shop.public.customers" db = sqlite3.connect("customer_projection.db") db.execute(""" CREATE TABLE IF NOT EXISTS customer_projection ( id INTEGER PRIMARY KEY, full_name TEXT NOT NULL, email TEXT NOT NULL, updated_at TEXT NOT NULL ) """) db.commit() consumer = KafkaConsumer( TOPIC, bootstrap_servers="localhost:9092", group_id="sqlite-customer-sink-v1", auto_offset_reset="earliest", enable_auto_commit=False, key_deserializer=lambda raw: json.loads(raw.decode("utf-8")) if raw else None, value_deserializer=lambda raw: json.loads(raw.decode("utf-8")) if raw else None, ) for message in consumer: key = message.key value = message.value if not key or "id" not in key: raise ValueError(f"Missing primary-key message key: {key}") customer_id = key["id"] is_delete = ( value is None or str(value.get("__deleted", "")).lower() == "true" ) with db: if is_delete: db.execute( "DELETE FROM customer_projection WHERE id = ?", (customer_id,) ) else: db.execute(""" INSERT INTO customer_projection (id, full_name, email, updated_at) VALUES (?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET full_name = excluded.full_name, email = excluded.email, updated_at = excluded.updated_at """, ( customer_id, value["full_name"], value["email"], value["updated_at"], )) # Offset is committed only after SQLite commits successfully.  consumer.commit()

Run it:

python customer_sink.py

If the consumer reads the initial snapshot followed by the three changes above, query the sink database:

sqlite3 customer_projection.db \ "SELECT id, full_name, email FROM customer_projection ORDER BY id;"

Expected result:

1|Aisha Rahman|aisha.rahman@example.com 3|Omar Haddad|omar@example.com

id=2 is absent because its source row was deleted.

The upsert makes repeated create, snapshot, and update events safe. The delete is also idempotent: deleting an already absent row is harmless.

Handle Ordering, Duplicates, and Schema Changes Deliberately

Kafka guarantees order only within one partition. Records with the same non-null key are normally routed to the same partition, so all changes for customer id=1 retain order.

Do not assume a global order across customers. An update to customer 1 and an insert for customer 3 may be processed in different partitions and in a different relative order than another system observes them.

Duplicates are normal under at-least-once delivery. The SQLite sample handles them because:

inserts are upserts;
updates overwrite the same primary key;
deletes are repeatable;
the consumer commits offsets after the sink write.

Schema evolution needs an explicit plan. Adding a nullable source column is usually straightforward, but dropping or renaming columns can break consumers. For production pipelines:

version event contracts;
deploy sink support before emitting a required new field;
avoid SELECT *-style assumptions in consumers;
use a schema registry and compatibility rules if multiple services consume the topic.

The JSON converter simplifies this tutorial, but it does not provide strong schema governance. Avro, Protobuf, or JSON Schema with a registry is often preferable for shared event streams.

Production Pitfalls and Operational Controls

A replication slot retains WAL until Debezium confirms it has consumed it. If Kafka Connect is down for too long, PostgreSQL disk usage can grow because old WAL segments cannot be removed. Monitor replication-slot lag and WAL disk usage.

Other common pitfalls include:

No primary key: deletes and updates become difficult to apply safely downstream.
Snapshot surprises: an initial snapshot can produce many records. Ensure the sink can handle the initial volume and that topic retention is sufficient.
Connector restarts: source offsets are stored by Kafka Connect. Losing Connect internal topics can cause unwanted re-snapshots or duplicate processing.
Poison messages: one malformed record can repeatedly fail a consumer. Production consumers should validate records, log context, and use a dead-letter strategy.
Delete semantics: decide whether sinks need hard deletes, soft deletes, tombstones, or historical retention.
Database permissions: grant Debezium only the replication and read permissions it needs.
Topic management: pre-create production topics with intentional partition counts, retention, compaction choices, ACLs, and monitoring.

For a current-state projection, consider log compaction on the CDC topic only after understanding delete tombstones and consumer replay requirements. A compacted topic is useful for rebuilding latest state, while a retained append-only topic is more useful for historical replay.

Test the Pipeline and Verify Recovery Behavior

Test CDC as a data contract, not just a connectivity check.

Start with an integration test environment containing PostgreSQL, Kafka, Kafka Connect, and a disposable sink. Then verify:

Initial snapshot: source rows appear in the sink.
Insert: a new source row appears once logically in the sink.
Update: changed values replace old values.
Delete: the sink removes or marks the row according to your contract.
Restart recovery: stop the sink after a database write but before an offset commit, then restart it. The duplicate event must not corrupt the result.
Connector recovery: restart Kafka Connect and confirm streaming resumes from stored offsets.
Schema change: add a nullable column and verify consumers tolerate it.

A useful assertion compares source and sink after all events have settled:

SELECT id, full_name, email FROM public.customers ORDER BY id;

Compare that result with the sink projection query. For this current-state sink, they should match on the columns intentionally replicated.

Takeaway

Debezium CDC turns committed PostgreSQL changes into Kafka events by reading the WAL, not by polling tables. A reliable source-to-sink pipeline depends on stable primary keys, idempotent sink writes, offset commits after durable writes, and monitoring of replication-slot lag.

The smallest useful mental model is: PostgreSQL commits a change, Debezium publishes it, Kafka retains it, and the sink repeatedly applies it safely until its state converges with the source.