Implementing Change Data Capture Pipelines with Debezium: Capturing and Routing Database Events in Production

Introduction

Imagine you're the data engineer responsible for keeping a downstream analytics warehouse in sync with a production PostgreSQL database that processes thousands of order updates per minute. Your current approach — a nightly batch job that pulls everything modified since midnight — has served you well enough, but now the business wants real-time inventory dashboards, fraud detection that fires in seconds, and audit trails that capture every intermediate state of an order. A nightly batch job can't give you any of that.

This is the problem Change Data Capture (CDC) was designed to solve. Instead of periodically querying your database and comparing results, CDC taps directly into the database's transaction log — the same mechanism the database uses for replication — and streams every insert, update, and delete as it happens. Debezium is the most production-proven open-source CDC framework for doing exactly this, and understanding how to configure, deploy, and operate it is a critical skill for any serious data engineer.

By the end of this lesson, you'll have built a complete, production-grade CDC pipeline that captures events from PostgreSQL, routes them through Kafka, and lands transformed records in a downstream sink. You'll understand not just how to configure Debezium connectors, but why the underlying architecture works the way it does — which is essential when something breaks at 2 AM.

What you'll learn:

• How Debezium reads from database transaction logs and what that means for your database configuration
• How to deploy and configure a Debezium PostgreSQL connector with production-appropriate settings
• How Kafka topic naming conventions and event schemas are structured by Debezium
• How to route, filter, and transform CDC events using Kafka Connect Single Message Transforms (SMTs)
• How to handle schema evolution, snapshotting, and connector failure recovery without data loss

---

Prerequisites

You should be comfortable with Kafka fundamentals (topics, producers, consumers, consumer groups) and have working knowledge of at least one relational database. If you're new to the broader pipeline landscape, it helps to first read the incremental loading patterns overview to understand where CDC fits relative to timestamp-based and watermark approaches. You should also have Docker available locally for the hands-on exercises.

---

How Debezium Actually Works: The Transaction Log Approach

Before you write a single line of configuration, you need to understand what Debezium is actually doing under the hood. This isn't academic — the architecture has direct implications for how you configure your database, what permissions your connector user needs, and what failure modes you'll encounter in production.

Every major relational database maintains a transaction log (called the Write-Ahead Log in PostgreSQL, the Binary Log in MySQL, or LogMiner in Oracle). This log is the authoritative record of every change that occurred in the database, written before the change is applied to the data files. The database's own replication system reads this log to propagate changes to replicas. Debezium does exactly the same thing: it presents itself to the database as a replication client and streams the log.

This approach has two massive advantages over polling-based CDC. First, it captures every change, including multiple updates to the same row within a single batch window and deletes (which are invisible to timestamp-based approaches). Second, it has near-zero impact on the database because reading the WAL doesn't require table scans.

The trade-off is that your database needs to retain the WAL long enough for Debezium to read it. If Debezium falls behind — because Kafka is slow, the connector is paused, or you're doing an initial snapshot — and the WAL segments rotate before they're consumed, you'll need to re-snapshot from scratch.

Key insight: Debezium's durability guarantee comes from Kafka, not the database. Once Debezium reads an event from the WAL and publishes it to Kafka, it stores the current WAL offset in a Kafka topic called connect-offsets. If the connector restarts, it picks up exactly where it left off — but only if those WAL segments still exist on the database side.

---

Configuring PostgreSQL for Logical Replication

PostgreSQL requires specific configuration to expose the information Debezium needs. You'll configure this once per PostgreSQL instance (not per connector), and it requires a database restart.

postgresql.conf Changes

# Enable logical replication
wal_level = logical

# How many replication slots can exist simultaneously
# Each Debezium connector uses one slot
max_replication_slots = 5

# How many WAL sender processes can run simultaneously
max_wal_senders = 5

# How long to retain WAL segments (in megabytes)
# Increase this if Debezium might fall behind during high-volume periods
wal_keep_size = 1024

Creating the Replication User

Never use your application's database user for Debezium. Create a dedicated user with exactly the permissions needed:

-- Create dedicated Debezium user
CREATE USER debezium_user WITH PASSWORD 'secure_password_here' REPLICATION LOGIN;

-- Grant read access to the tables you want to capture
GRANT SELECT ON ALL TABLES IN SCHEMA public TO debezium_user;
ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO debezium_user;

-- Grant access to the replication catalog
GRANT USAGE ON SCHEMA public TO debezium_user;

-- If using PostgreSQL 14+, grant pg_monitor for health checks
GRANT pg_monitor TO debezium_user;

Setting Up the Replication Slot

Debezium can create the replication slot automatically on first start, but in production you often want to create it manually so you can verify it before deploying the connector:

-- Create a replication slot using the pgoutput plugin (Postgres 10+)
SELECT pg_create_logical_replication_slot('debezium_orders_slot', 'pgoutput');

-- Verify it was created
SELECT slot_name, plugin, slot_type, active, restart_lsn, confirmed_flush_lsn 
FROM pg_replication_slots;

Warning: Replication slots prevent WAL segments from being deleted until the slot has consumed them. If your Debezium connector is paused or fails for an extended period, these retained WAL segments will accumulate and can fill your disk. Monitor the pg_replication_slots view and set up alerts when pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn) grows large. A reasonable alert threshold is 5GB.

Table Publication

PostgreSQL's logical replication uses a concept called a "publication" — an explicit declaration of which tables to replicate:

-- Publish all tables in the schema
CREATE PUBLICATION debezium_publication FOR ALL TABLES;

-- Or, more safely in production, publish only specific tables
CREATE PUBLICATION debezium_publication FOR TABLE 
  orders, 
  order_items, 
  customers, 
  inventory;

---

Deploying Debezium with Kafka Connect

Debezium runs as a Kafka Connect plugin. You deploy Kafka Connect workers, and Debezium connectors run inside those workers. Here's a complete Docker Compose setup for local development:

version: '3.8'

services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.4.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  kafka:
    image: confluentinc/cp-kafka:7.4.0
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_LOG_RETENTION_HOURS: 168
      KAFKA_LOG_SEGMENT_BYTES: 1073741824

  schema-registry:
    image: confluentinc/cp-schema-registry:7.4.0
    depends_on:
      - kafka
    ports:
      - "8081:8081"
    environment:
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: kafka:9092

  kafka-connect:
    image: debezium/connect:2.4
    depends_on:
      - kafka
      - schema-registry
    ports:
      - "8083:8083"
    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.connect.avro.AvroConverter
      CONNECT_VALUE_CONVERTER: io.confluent.connect.avro.AvroConverter
      CONNECT_KEY_CONVERTER_SCHEMA_REGISTRY_URL: http://schema-registry:8081
      CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL: http://schema-registry:8081

Note: The Debezium Docker image (debezium/connect) comes with Debezium connectors pre-installed. If you're using a plain Confluent Kafka Connect image, you need to install the Debezium connector JARs separately, either by extending the Docker image or by mounting them as a volume.

---

Configuring the Debezium PostgreSQL Connector

With infrastructure running, you register a connector by POSTing a JSON configuration to the Kafka Connect REST API. Here's a production-appropriate configuration for capturing order events:

{
  "name": "postgres-orders-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "plugin.name": "pgoutput",
    
    "database.hostname": "your-postgres-host.internal",
    "database.port": "5432",
    "database.user": "debezium_user",
    "database.password": "${file:/kafka/external-secrets.properties:db.password}",
    "database.dbname": "production",
    "database.server.name": "prod-orders",
    
    "slot.name": "debezium_orders_slot",
    "publication.name": "debezium_publication",
    
    "table.include.list": "public.orders,public.order_items,public.customers",
    
    "snapshot.mode": "initial",
    "snapshot.locking.mode": "none",
    
    "heartbeat.interval.ms": "30000",
    "heartbeat.action.query": "INSERT INTO debezium_heartbeat (ts) VALUES (NOW()) ON CONFLICT (id) DO UPDATE SET ts = EXCLUDED.ts",
    
    "decimal.handling.mode": "double",
    "time.precision.mode": "connect",
    
    "topic.prefix": "prod",
    
    "errors.tolerance": "all",
    "errors.log.enable": "true",
    "errors.log.include.messages": "true",
    "errors.deadletterqueue.topic.name": "prod.dlq.orders-connector",
    "errors.deadletterqueue.topic.replication.factor": "3",
    
    "max.batch.size": "2048",
    "max.queue.size": "16384",
    "poll.interval.ms": "500"
  }
}

Deploy this with a curl command:

curl -X POST \
  http://localhost:8083/connectors \
  -H 'Content-Type: application/json' \
  -d @postgres-orders-connector.json

Verify it's running:

curl http://localhost:8083/connectors/postgres-orders-connector/status | jq .

A healthy response looks like:

{
  "name": "postgres-orders-connector",
  "connector": {
    "state": "RUNNING",
    "worker_id": "kafka-connect:8083"
  },
  "tasks": [
    {
      "id": 0,
      "state": "RUNNING",
      "worker_id": "kafka-connect:8083"
    }
  ]
}

Understanding the Topic Naming Convention

Debezium creates Kafka topics automatically, following this pattern:

{topic.prefix}.{schema_name}.{table_name}

With our configuration, you'll see:

• prod.public.orders
• prod.public.order_items
• prod.public.customers

This matters for your consumers and for configuring downstream connectors. If you're routing to multiple environments, making topic.prefix environment-specific (like prod, staging) keeps topics cleanly separated.

---

Understanding the Debezium Event Schema

Every event Debezium publishes has a structured envelope. Understanding this structure is essential for writing consumers and transforms. Here's what an order update event looks like:

{
  "schema": { ... },
  "payload": {
    "before": {
      "order_id": 98234,
      "customer_id": 1042,
      "status": "pending",
      "total_amount": 149.99,
      "updated_at": 1698765432000000
    },
    "after": {
      "order_id": 98234,
      "customer_id": 1042,
      "status": "confirmed",
      "total_amount": 149.99,
      "updated_at": 1698765498000000
    },
    "source": {
      "version": "2.4.0.Final",
      "connector": "postgresql",
      "name": "prod",
      "ts_ms": 1698765498123,
      "snapshot": "false",
      "db": "production",
      "sequence": "[\"1234567890\",\"1234567891\"]",
      "schema": "public",
      "table": "orders",
      "txId": 7823,
      "lsn": 1234567891,
      "xmin": null
    },
    "op": "u",
    "ts_ms": 1698765498456,
    "transaction": {
      "id": "7823:1234567891",
      "total_order": 2,
      "data_collection_order": 1
    }
  }
}

The op field is your primary routing signal:

• c — create (INSERT)
• u — update (UPDATE)
• d — delete (DELETE)
• r — read (during snapshot)

The before field contains the row state before the change. For inserts, it's null. For deletes, after is null. This "before and after" structure is what enables downstream systems to implement idempotent upsert semantics — you always know both what changed and what it changed from.

The source.lsn field (Log Sequence Number) is your authoritative ordering marker. Events from the same transaction share a txId and are guaranteed to arrive in order within that transaction.

Key insight: The ts_ms at the top level of the payload is when Debezium processed the event — not when the database change occurred. Use source.ts_ms for the actual database transaction timestamp. This distinction matters enormously for time-series analytics and event-time windowing in streaming consumers.

---

Routing and Transforming Events with Single Message Transforms

Raw Debezium events are powerful but verbose. Most downstream consumers don't need the full envelope — a fraud detection service just wants the new order status and customer ID, not the before state and source metadata. Kafka Connect's Single Message Transform (SMT) framework lets you reshape events in the connector itself, before they hit Kafka.

Extracting Just the "After" State

The most common transform is ExtractNewRecordState, which unwraps the Debezium envelope and produces a flat record:

{
  "transforms": "unwrap,addMetadata",
  
  "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
  "transforms.unwrap.drop.tombstones": "false",
  "transforms.unwrap.delete.handling.mode": "rewrite",
  "transforms.unwrap.add.fields": "op,source.ts_ms,source.txId",
  "transforms.unwrap.add.headers": "op",
  
  "transforms.addMetadata.type": "org.apache.kafka.connect.transforms.InsertField$Value",
  "transforms.addMetadata.static.field": "pipeline_version",
  "transforms.addMetadata.static.value": "v2.4"
}

With ExtractNewRecordState, your flattened order update now looks like:

{
  "order_id": 98234,
  "customer_id": 1042,
  "status": "confirmed",
  "total_amount": 149.99,
  "updated_at": 1698765498000000,
  "__op": "u",
  "__source_ts_ms": 1698765498123,
  "__source_txId": 7823,
  "pipeline_version": "v2.4"
}

Much cleaner for downstream consumers, and the __op prefix lets you still distinguish inserts from updates when needed.

Routing Events to Different Topics by Operation Type

For some architectures — say, you have one service that handles new orders and another that handles order updates — you want to split by op. Use the TopicNameMatches and Filter SMTs, or the ContentBasedRouter from Debezium:

{
  "transforms": "unwrap,routeByOp",
  
  "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
  "transforms.unwrap.add.fields": "op",
  
  "transforms.routeByOp.type": "io.debezium.transforms.ContentBasedRouter",
  "transforms.routeByOp.language": "jsr223.groovy",
  "transforms.routeByOp.topic.expression": 
    "value.__op == 'c' ? 'prod.orders.new' : value.__op == 'd' ? 'prod.orders.deleted' : null"
}

Events with op == 'u' stay on the original topic; inserts go to prod.orders.new and deletes to prod.orders.deleted. This fan-out pattern connects naturally to the data pipeline design patterns around fan-out and branching workflows.

Filtering Out Irrelevant Events

If you're capturing a high-volume table but only care about certain state transitions — for example, only orders that transition to shipped or cancelled status:

{
  "transforms": "unwrap,filterByStatus",
  
  "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
  "transforms.unwrap.add.fields": "op",
  
  "transforms.filterByStatus.type": "io.debezium.transforms.Filter",
  "transforms.filterByStatus.language": "jsr223.groovy",
  "transforms.filterByStatus.condition": 
    "value.__op != 'd' && (value.status == 'shipped' || value.status == 'cancelled')"
}

Warning: SMT-based filtering happens after events are read from the WAL but before they're published to Kafka. Filtered events are silently dropped — they never appear in any topic. If you later discover your filter condition was wrong, those events are gone. Consider publishing unfiltered events to a raw topic first and filtering in your consumers, or use Kafka Streams / ksqlDB for stateful filtering downstream.

---

Handling Snapshots in Production

When you first start a Debezium connector against a live database, it needs to establish a baseline. By default (snapshot.mode: initial), it reads every row in your included tables before switching to streaming mode. For large tables, this can take hours.

Snapshot Modes

• initial — Full table scan on first start, then streaming. This is the safe default.
• initial_only — Snapshot then stop. Useful for one-time migrations.
• never — Skip snapshot entirely, start streaming from the current WAL position. Use this when you know your downstream already has current data (e.g., you restored from a database backup and just want forward changes).
• always — Re-snapshot on every connector start. Rarely appropriate in production.
• exported — Snapshot using an existing consistent export. Useful for very large databases.

For tables with tens of millions of rows, configure snapshot chunking:

{
  "snapshot.fetch.size": "10240",
  "snapshot.max.threads": "4",
  "snapshot.select.statement.overrides": "public.orders:SELECT * FROM public.orders WHERE created_at > NOW() - INTERVAL '2 years'"
}

The snapshot.select.statement.overrides lets you snapshot a filtered subset of the table — here, only orders from the last two years — while still capturing all future changes via streaming. This dramatically reduces snapshot time for tables with large historical data you don't need downstream.

Tip: During snapshot, Debezium sets op to r (read) instead of c (create). If your downstream consumer is using op to decide between INSERT and UPSERT, make sure it handles r the same way it handles c. The ExtractNewRecordState SMT can help here: set "transforms.unwrap.handle.deletes": "rewrite" and ensure your consumer treats all non-delete events as upserts.

---

Schema Evolution with Debezium

Production databases change. Columns get added, renamed, or dropped. When the schema changes in PostgreSQL, Debezium detects it via the WAL and automatically updates the Avro schema in the Schema Registry. But not all changes are created equal.

Adding a nullable column is generally safe — Debezium will include the new column in subsequent events, and older Avro schemas can be made forward-compatible with a default of null. Dropping a column or changing a column's type is a breaking change that requires coordination.

The detailed strategies for managing these changes without downtime are covered in depth in the schema evolution strategies guide, but here's the Debezium-specific workflow for adding a column safely:

1. Add the column to PostgreSQL as nullable with a default value
2. Verify the new Avro schema registered in Schema Registry is backward-compatible
3. Update your consumer to handle both the old schema (events before the column was added) and the new schema
4. Deploy consumer changes before the database migration runs

For column renames or type changes, use a two-phase approach: add the new column alongside the old one, migrate data, update consumers to read from the new column, then drop the old column in a later deployment.

---

Building a Complete Sink: PostgreSQL CDC to ClickHouse

Let's put this all together with a realistic scenario: streaming order events from a production PostgreSQL OLTP database into ClickHouse for real-time analytics. We'll use a Python consumer that reads from Kafka and writes to ClickHouse.

from confluent_kafka import Consumer, KafkaError
from confluent_kafka.avro import AvroConsumer
from confluent_kafka.avro.serializer import SerializerError
import clickhouse_connect
import json
import logging
from datetime import datetime
from typing import Optional

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class OrderEventProcessor:
    def __init__(self, kafka_config: dict, clickhouse_config: dict):
        self.consumer = AvroConsumer({
            **kafka_config,
            'group.id': 'orders-to-clickhouse-v1',
            'auto.offset.reset': 'earliest',
            'enable.auto.commit': False,  # Manual commit for exactly-once semantics
            'max.poll.interval.ms': 300000,
        })
        self.consumer.subscribe(['prod.public.orders'])
        
        self.ch_client = clickhouse_connect.get_client(**clickhouse_config)
        self.batch = []
        self.batch_size = 500
        self.flush_interval_seconds = 10
        self.last_flush = datetime.now()
        
    def transform_event(self, value: dict) -> Optional[dict]:
        """Transform a flattened Debezium event for ClickHouse."""
        op = value.get('__op')
        
        # We handle creates, updates, and snapshot reads as upserts
        # Deletes use ClickHouse's ReplacingMergeTree with a is_deleted flag
        if op == 'd':
            return {
                'order_id': value['order_id'],
                'customer_id': value.get('customer_id'),
                'status': 'deleted',
                'total_amount': 0.0,
                'event_time': datetime.fromtimestamp(value['__source_ts_ms'] / 1000),
                'is_deleted': 1,
                'pipeline_version': 'v2.4',
            }
        
        return {
            'order_id': value['order_id'],
            'customer_id': value['customer_id'],
            'status': value['status'],
            'total_amount': float(value.get('total_amount', 0)),
            'event_time': datetime.fromtimestamp(value['__source_ts_ms'] / 1000),
            'is_deleted': 0,
            'pipeline_version': 'v2.4',
        }
    
    def flush_batch(self):
        """Flush accumulated events to ClickHouse."""
        if not self.batch:
            return
            
        try:
            self.ch_client.insert(
                'analytics.orders',
                self.batch,
                column_names=[
                    'order_id', 'customer_id', 'status', 
                    'total_amount', 'event_time', 'is_deleted', 'pipeline_version'
                ]
            )
            self.consumer.commit()
            logger.info(f"Flushed {len(self.batch)} events to ClickHouse")
            self.batch = []
            self.last_flush = datetime.now()
        except Exception as e:
            logger.error(f"Failed to flush batch: {e}")
            # Don't commit offset — events will be reprocessed
            raise
    
    def should_flush(self) -> bool:
        elapsed = (datetime.now() - self.last_flush).seconds
        return len(self.batch) >= self.batch_size or elapsed >= self.flush_interval_seconds
    
    def run(self):
        logger.info("Starting order event processor")
        
        try:
            while True:
                try:
                    msg = self.consumer.poll(timeout=1.0)
                except SerializerError as e:
                    logger.error(f"Deserialization error: {e}")
                    # Route to dead letter queue handling
                    continue
                
                if msg is None:
                    if self.should_flush():
                        self.flush_batch()
                    continue
                
                if msg.error():
                    if msg.error().code() == KafkaError._PARTITION_EOF:
                        continue
                    logger.error(f"Kafka error: {msg.error()}")
                    continue
                
                transformed = self.transform_event(msg.value())
                if transformed:
                    self.batch.append(transformed)
                
                if self.should_flush():
                    self.flush_batch()
                    
        except KeyboardInterrupt:
            logger.info("Shutting down — flushing remaining events")
            self.flush_batch()
        finally:
            self.consumer.close()


if __name__ == '__main__':
    processor = OrderEventProcessor(
        kafka_config={
            'bootstrap.servers': 'kafka:9092',
            'schema.registry.url': 'http://schema-registry:8081',
        },
        clickhouse_config={
            'host': 'clickhouse.internal',
            'port': 8123,
            'username': 'analytics_writer',
            'password': 'secure_password',
        }
    )
    processor.run()

The corresponding ClickHouse table uses ReplacingMergeTree, which deduplicates rows by order_id and keeps the version with the latest event_time:

CREATE TABLE analytics.orders
(
    order_id      UInt64,
    customer_id   UInt64,
    status        LowCardinality(String),
    total_amount  Float64,
    event_time    DateTime,
    is_deleted    UInt8,
    pipeline_version String
)
ENGINE = ReplacingMergeTree(event_time)
ORDER BY order_id
PARTITION BY toYYYYMM(event_time);

For queries that need the current state of each order, use FINAL:

SELECT order_id, customer_id, status, total_amount
FROM analytics.orders FINAL
WHERE is_deleted = 0
  AND status = 'confirmed'
  AND event_time >= toStartOfDay(now())
ORDER BY order_id;

This is also where data quality validation becomes critical — you should validate that record counts between PostgreSQL and ClickHouse converge within acceptable drift thresholds. A nightly reconciliation job comparing COUNT(*) per status between the two systems catches drift before it becomes a business problem.

---

Hands-On Exercise

In this exercise, you'll build a CDC pipeline that captures customer table changes and routes high-value customer updates to a dedicated Kafka topic for a marketing automation service.

Setup: Clone the following starter docker-compose and seed your database:

# Start infrastructure
docker compose up -d

# Wait for services to be healthy, then seed
docker exec -it postgres psql -U postgres -d production << 'EOF'
CREATE TABLE customers (
    customer_id SERIAL PRIMARY KEY,
    email VARCHAR(255) UNIQUE NOT NULL,
    tier VARCHAR(50) DEFAULT 'standard',
    lifetime_value NUMERIC(12,2) DEFAULT 0.00,
    created_at TIMESTAMP DEFAULT NOW(),
    updated_at TIMESTAMP DEFAULT NOW()
);

INSERT INTO customers (email, tier, lifetime_value) VALUES
    ('alice@example.com', 'standard', 450.00),
    ('bob@example.com', 'standard', 89.50),
    ('carol@example.com', 'premium', 12400.00);
EOF

Step 1: Register the connector via the REST API. Create a file customer-connector.json with:

• Source: the customers table
• topic.prefix: prod
• snapshot.mode: initial
• Add the ExtractNewRecordState transform
• Add a ContentBasedRouter transform that routes events where lifetime_value > 1000 to prod.customers.high-value

Step 2: Verify events are landing in Kafka:

docker exec -it kafka kafka-console-consumer \
  --bootstrap-server localhost:9092 \
  --topic prod.public.customers \
  --from-beginning \
  --max-messages 10

Step 3: Update one of the standard customers to cross the $1,000 threshold:

UPDATE customers 
SET lifetime_value = 1500.00, updated_at = NOW() 
WHERE email = 'bob@example.com';

Step 4: Verify the update event appears in prod.customers.high-value but not in prod.public.customers (or vice versa depending on your routing logic). Use kafka-console-consumer with --from-beginning on each topic.

Step 5: Write a Python consumer that reads from prod.customers.high-value and prints a formatted marketing trigger message: "High-value customer upgrade: {email} now at ${lifetime_value}. Trigger VIP onboarding flow."

---

Common Mistakes & Troubleshooting

The Connector Keeps Restarting

Check connector status first:

curl http://localhost:8083/connectors/postgres-orders-connector/status | jq '.tasks[0].trace'

Common causes:

• Replication slot doesn't exist: You deleted the slot manually but the connector is trying to use it. Either recreate the slot or change slot.name in the connector config.
• WAL segments rotated past the slot's position: The restart_lsn in pg_replication_slots is ahead of the slot's confirmed_flush_lsn. You need to delete the slot, drop and recreate the connector, and accept a re-snapshot.
• Database password expired: Create replication users with VALID UNTIL 'infinity'.

Events Are Out of Order Across Tables

Debezium guarantees ordering within a single table's topic partition, but events from orders and order_items arrive in separate topics with no cross-topic ordering guarantee. If you need to reconstruct transactions (e.g., an order and its line items together), use the transaction.topic feature:

{
  "provide.transaction.metadata": "true"
}

This enables a {topic.prefix}.transaction topic where Debezium publishes transaction boundaries, letting you group related events by txId. For consuming transactional groups correctly, reference the checkpointing and state management patterns for handling partial transaction windows.

Snapshot Is Taking Too Long

If your snapshot is running for hours and blocking deployment:

1. Increase snapshot.fetch.size to reduce round trips (try 10,000–50,000 rows per fetch)
2. Use snapshot.select.statement.overrides to limit which rows are snapshotted
3. Check if snapshot.locking.mode: none is set — without this, Debezium takes an exclusive lock during snapshot on older PostgreSQL versions
4. Consider parallelizing with snapshot.max.threads but be careful of I/O impact on the production database

Consumer Lag Is Growing

If your Kafka consumer group lag is increasing, you have a throughput mismatch. The backpressure and throughput tuning guide covers the full diagnosis process, but for Debezium-specific issues:

• Increase max.batch.size and max.queue.size in the connector config
• If the sink is the bottleneck, scale your consumer group by adding more consumer instances (Kafka will rebalance partitions)
• If the source is producing too fast, implement circuit breaker patterns on the sink side to apply backpressure gracefully

Dead Letter Queue Is Filling Up

If events are hitting your DLQ topic, inspect them:

kafka-console-consumer \
  --bootstrap-server localhost:9092 \
  --topic prod.dlq.orders-connector \
  --from-beginning \
  --property print.headers=true

The headers contain the error reason. Common causes:

• Schema incompatibility (a DDL change you didn't handle)
• Null values in non-nullable schema fields
• Deserialization errors from malformed WAL entries (rare but possible during PostgreSQL upgrades)

For a comprehensive framework for handling these cases, the dead letter queue implementation guide covers replayability, alerting, and human-in-the-loop review workflows.

Tip: Set up an alert that fires when your DLQ topic's consumer lag (from a monitoring consumer) exceeds 100 messages. A growing DLQ is almost always a symptom of a systemic problem — schema drift, a new data pattern the connector isn't handling — not a series of isolated bad records. Pipeline observability practices should include DLQ lag as a first-class metric alongside consumer group lag and connector status.

---

Summary & Next Steps

You've now built a complete understanding of Debezium CDC pipelines: from configuring PostgreSQL's WAL and replication slots, to deploying Kafka Connect connectors with production-appropriate settings, to routing and transforming events with SMTs, to writing consumers that handle the full event lifecycle including deletes and schema snapshots.

The key architectural principles to carry forward:

• Debezium's durability is Kafka's durability. Once in Kafka, events are safe. The WAL is transient — monitor replication slot lag to ensure you're consuming fast enough.
• The event envelope is your audit trail. The before/after structure plus source.lsn and txId give you complete provenance for every change.
• SMTs are powerful but irreversible. Filter and transform conservatively; prefer keeping raw events alongside transformed ones.
• Snapshots are one-time costs. Design your snapshot strategy before deploying to production — a poorly configured snapshot on a 500GB table is painful to abort and restart.

Where to go from here:

• Explore MySQL and SQL Server CDC with Debezium — the connector configs differ significantly (binlog vs. LogMiner) but the Kafka-side architecture is identical
• Implement idempotent pipeline semantics in your consumers so that reprocessing CDC events never creates duplicate records
• Study slowly changing dimension patterns to understand how CDC events map to Type 1, Type 2, and Type 3 dimension updates in a warehouse
• Add pipeline-level observability with connector health checks, consumer group lag dashboards, and DLQ alerting — in production, you'll know your CDC pipeline is healthy not by checking logs, but by dashboards that haven't fired an alert in days