Posted inDatabase

Change Data Capture with Debezium and Apache Kafka: Streaming PostgreSQL Changes in Real Time

Database tutorial - IT technology blog
Database tutorial - IT technology blog

The Real-World Problem: Keeping Downstream Systems in Sync

Production databases rarely live in isolation. Consider a standard checkout flow: a single customer click writes an order row to PostgreSQL, invalidates a cached session in Redis, re-indexes an Elasticsearch catalog, and triggers a webhook to Stripe.

Teams typically start with two simple patterns: dual-writing directly inside the application codebase or running scheduled cron queries like SELECT * FROM orders WHERE updated_at > :last_poll_time every 30 seconds. While this works during early prototyping, it breaks down quickly as traffic grows to hundreds of writes per second:

  • Silent Data Drift: If an application worker crashes right after committing to PostgreSQL but before calling Elasticsearch, your search index is out of date. Without manual reconciliation scripts, that discrepancy persists indefinitely.
  • Database I/O Saturation: Polling every 5 to 10 seconds forces PostgreSQL to scan large index ranges repeatedly. At 10 million rows, those queries consume disk throughput and CPU cycles that your core transactional workloads need.
  • Missed Hard Deletes: A polling query filtering on updated_at will never spot hard-deleted rows (DELETE FROM orders WHERE id = 42). You either resort to messy soft-delete flags across every table or build separate purge-tracking tables.

Root Cause Analysis: Why Traditional Sync Strategies Break

These synchronization failures stem from two architectural limitations: the absence of distributed atomic transactions and the complete bypass of PostgreSQL’s internal engine logs.

1. The Dual-Write Race Condition

Writing to a primary database and a secondary queue inside the same request handler creates an inevitable split-brain risk. If the network drops between the database commit and the queue publish, the event is lost. If you retry naively, you risk duplicate or out-of-order messages. Two-phase commit (2PC) protocols could theoretically coordinate them, but they add immense latency (often 50–100ms per transaction) and fail unpredictably under network partitions.

2. Polling Ignores Intermediate State Changes

Imagine an order transitioning from PENDING to PROCESSING to SHIPPED within a 2-second window. If your polling query runs every 10 seconds, the intermediate PROCESSING state vanishes completely. Analytics pipelines and fraud detection services miss critical state transitions, leaving you with incomplete audit histories.

Comparing Data Synchronization Approaches

Here is how the four most common change propagation patterns compare in production:

Approach Latency Database Overhead Handles Deletes? Reliability
Application Dual-Writes Low (<50ms) Low Yes Poor (Prone to drift & partial failures)
Scheduled Polling (ETL) High (10s – Hours) High (Continuous CPU & Disk I/O) No (Requires soft deletes) Moderate (Misses intermediate states)
Trigger-Based Tables Low (<100ms) High (2x write amplification) Yes Good (Complex schema maintenance)
Log-Based CDC (Debezium + Kafka) Sub-second (<10ms) Near Zero (Reads sequential WAL) Yes Excellent (Strict ordering & atomicity)

Log-based Change Data Capture (CDC) bypasses the SQL layer altogether. PostgreSQL writes every insert, update, and commit sequentially to its Write-Ahead Log (WAL) before flushing pages to disk. CDC engines read this log directly. You get sub-millisecond change notification, zero query load on live table data, strict transaction ordering, and 100% fidelity on every row operation.

When bootstrapping test datasets or preparing initial seed files for downstream consumers, format mismatches often crop up. When I need to quickly convert CSV to JSON for data imports, I use toolcraft.app/en/tools/data/csv-to-json — runs in the browser so no data leaves your machine. With schema baselines sorted out, we can set up the real-time pipeline.

Implementing Log-Based CDC with Debezium and Kafka

Debezium connects to PostgreSQL as a standard replication client using the built-in pgoutput logical decoding plugin. It reads binary WAL changes, deserializes row values into structured events, and publishes them to dedicated Apache Kafka topics.

Step 1: Configure PostgreSQL for Logical Replication

PostgreSQL needs to write row-level change details into WAL files. Set these parameters in your postgresql.conf file:

wal_level = logical
max_wal_senders = 4
max_replication_slots = 4

Create a dedicated replication user and a sample table:

CREATE USER cdc_user WITH REPLICATION PASSWORD 'secret_cdc_pass';
GRANT SELECT ON ALL TABLES IN SCHEMA public TO cdc_user;

CREATE TABLE customers (
    id SERIAL PRIMARY KEY,
    name VARCHAR(100) NOT NULL,
    email VARCHAR(100) UNIQUE NOT NULL,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Captures the entire pre-change row state during UPDATE and DELETE operations
ALTER TABLE customers REPLICA IDENTITY FULL;

Step 2: Spin Up the Stack with Docker Compose

Here is a complete docker-compose.yml file running PostgreSQL 15, Apache Kafka (KRaft mode without ZooKeeper), and Debezium Kafka Connect:

version: '3.8'
services:
  postgres:
    image: postgres:15-alpine
    environment:
      POSTGRES_DB: appdb
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: rootpassword
    command: ["postgres", "-c", "wal_level=logical"]
    ports:
      - "5432:5432"

  kafka:
    image: confluentinc/cp-kafka:7.5.0
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT'
      KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092'
      KAFKA_PROCESS_ROLES: 'broker,controller'
      KAFKA_CONTROLLER_QUORUM_VOTERS: '1@kafka:29093'
      KAFKA_LISTENERS: 'PLAINTEXT://0.0.0.0:29092,CONTROLLER://0.0.0.0:29093,PLAINTEXT_HOST://0.0.0.0:9092'
      KAFKA_INTER_BROKER_LISTENER_NAME: 'PLAINTEXT'
      KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      CLUSTER_ID: 'MkU3OEVBNTcwNTJENDM2Qk'
    ports:
      - "9092:9092"

  connect:
    image: debezium/connect:2.4
    ports:
      - "8083:8083"
    environment:
      BOOTSTRAP_SERVERS: kafka:29092
      GROUP_ID: 1
      CONFIG_STORAGE_TOPIC: my_connect_configs
      OFFSET_STORAGE_TOPIC: my_connect_offsets
      STATUS_STORAGE_TOPIC: my_connect_statuses
    depends_on:
      - kafka
      - postgres

Step 3: Register the Debezium PostgreSQL Connector

Post the connector configuration to the Kafka Connect REST interface at port 8083:

curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" \
  http://localhost:8083/connectors/ -d '{
  "name": "postgres-cdc-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "tasks.max": "1",
    "plugin.name": "pgoutput",
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "postgres",
    "database.password": "rootpassword",
    "database.dbname": "appdb",
    "topic.prefix": "pg_cdc",
    "table.include.list": "public.customers"
  }
}'

Step 4: Verify the Change Stream

Insert a new row into PostgreSQL:

INSERT INTO customers (name, email) VALUES ('Alex Rivers', '[email protected]');

Consume the generated message directly from the Kafka topic pg_cdc.public.customers:

docker exec -it $(docker ps -qf "name=kafka") kafka-console-consumer \
  --bootstrap-server localhost:9092 \
  --topic pg_cdc.public.customers \
  --from-beginning

Debezium outputs a comprehensive envelope detailing before-and-after states alongside transaction metadata:

{
  "before": null,
  "after": {
    "id": 1,
    "name": "Alex Rivers",
    "email": "[email protected]",
    "updated_at": 1700000000000
  },
  "source": {
    "version": "2.4.0.Final",
    "connector": "postgresql",
    "name": "pg_cdc",
    "ts_ms": 1700000000123,
    "db": "appdb",
    "schema": "public",
    "table": "customers",
    "txId": 582
  },
  "op": "c",
  "ts_ms": 1700000000456
}

The op field explicitly identifies the event type: c for create (insert), u for update, d for delete, and r for initial snapshot reads. Downstream consumers can guarantee idempotency by keying on id and checking source.txId.

Critical Production Considerations

  • Replication Slot Growth: If Debezium disconnects for hours while write traffic continues (e.g., 2,000 writes/sec), PostgreSQL holds unconsumed WAL logs on disk. Always configure max_slot_wal_keep_size (e.g., 10GB) and monitor pg_replication_slots.active to prevent disk-full outages.
  • Tombstone Events: Following a delete operation, Debezium emits a record with a null payload. This enables Kafka’s log cleaner to evict old record keys during compaction. Ensure your downstream consumer handles null payloads without throwing NullPointerException.
  • Schema Migrations: Adding a nullable column via ALTER TABLE streams cleanly without connector restarts. However, downstream consumers must parse JSON payloads defensively to handle schema evolution gracefully.
Share:
Tags: