CDC Pipelines with Debezium: Production Patterns That Actually Hold Up

by

Most CDC setups work fine in dev. Then you push them to production, leave them running for three weeks, and wake up to a WAL slot that’s eaten 200GB of disk, a connector that silently stopped emitting events, and a changelog topic with a broken schema that nobody noticed. This article is about not letting that happen.

Debezium is genuinely good software. The problem isn’t Debezium — it’s that most tutorials stop at "connector running, events flowing" and skip straight to the congratulations. Production CDC has a completely different set of failure modes, and this guide covers the patterns that actually matter when real traffic hits.

The official repo and docs live at https://github.com/debezium/debezium. Read the connector-specific docs for your database; they’re unusually complete and save hours of trial-and-error.

What Debezium Is Actually Doing

Before diving into configs, a quick mental model check — because misunderstanding the mechanics leads directly to bad production decisions.

Debezium doesn’t poll your database. It taps into the database’s own replication mechanism: PostgreSQL logical replication slots, MySQL binlog, MongoDB change streams, SQL Server CDC tables. This is the source of both its power and its most dangerous failure modes.

For PostgreSQL specifically: Debezium creates a replication slot. The slot makes the WAL (Write-Ahead Log) durable — Postgres will never discard WAL segments that the slot hasn’t consumed yet. This is a feature, not a bug, until the consumer falls behind or dies, at which point your WAL grows unbounded and the database runs out of disk. That’s the number-one production incident with Postgres CDC.

For MySQL: the connector reads the binlog. You need binlog_format=ROW and the binlog needs to be retained long enough to survive any planned downtime of the connector. The default retention on managed MySQL instances is often 24-48h, which is dangerously short for anything serious.

The Stack

This article uses Kafka as the transport. Debezium also supports Pulsar and its standalone (embedded) server mode, but the Kafka Connect deployment model is the production standard for a reason — it has built-in horizontal scaling, offset management baked into Kafka itself, and mature operational tooling.

# docker-compose.yml
version: "3.8"

services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.6.1
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
    volumes:
      - zk-data:/var/lib/zookeeper/data
      - zk-log:/var/lib/zookeeper/log

  kafka:
    image: confluentinc/cp-kafka:7.6.1
    depends_on: [zookeeper]
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      # Keep logs long enough to survive planned outages
      KAFKA_LOG_RETENTION_HOURS: 168
      KAFKA_LOG_RETENTION_BYTES: 10737418240  # 10GB per partition
    volumes:
      - kafka-data:/var/lib/kafka/data

  schema-registry:
    image: confluentinc/cp-schema-registry:7.6.1
    depends_on: [kafka]
    ports:
      - "8081:8081"
    environment:
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: kafka:29092

  kafka-connect:
    image: debezium/connect:2.7
    depends_on: [kafka, schema-registry]
    ports:
      - "8083:8083"
    environment:
      BOOTSTRAP_SERVERS: kafka:29092
      GROUP_ID: debezium-connect-cluster
      CONFIG_STORAGE_TOPIC: debezium-configs
      OFFSET_STORAGE_TOPIC: debezium-offsets
      STATUS_STORAGE_TOPIC: debezium-status
      # Use Avro converters backed by Schema Registry
      KEY_CONVERTER: io.confluent.connect.avro.AvroConverter
      VALUE_CONVERTER: io.confluent.connect.avro.AvroConverter
      KEY_CONVERTER_SCHEMA_REGISTRY_URL: http://schema-registry:8081
      VALUE_CONVERTER_SCHEMA_REGISTRY_URL: http://schema-registry:8081
      # Tune connect worker for throughput
      OFFSET_FLUSH_INTERVAL_MS: 10000
    volumes:
      - connect-data:/kafka/connect

volumes:
  zk-data:
  zk-log:
  kafka-data:
  connect-data:

A note on converters: JSON converters are fine for exploration. Use Avro (or Protobuf) with Schema Registry in production. The schema evolution story is just better, and downstream consumers get typed deserialization for free.

PostgreSQL Connector: The Production Config

First, the database side. Debezium needs a replication role and the wal2json or pgoutput plugin. pgoutput is built into Postgres 10+ and preferred:

-- Run as superuser
CREATE ROLE debezium WITH REPLICATION LOGIN PASSWORD 'strong-password-here';
GRANT SELECT ON ALL TABLES IN SCHEMA public TO debezium;
ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO debezium;

-- pgoutput requires a publication
CREATE PUBLICATION debezium_pub FOR ALL TABLES;

Then the connector registration — POST this to https://cd-linux.club:8083/connectors:

{
  "name": "postgres-orders-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "strong-password-here",
    "database.dbname": "orders_db",
    "database.server.name": "orders",
    "plugin.name": "pgoutput",
    "publication.name": "debezium_pub",

    "table.include.list": "public.orders,public.order_items",

    "slot.name": "debezium_orders",
    "slot.drop.on.stop": "false",

    "snapshot.mode": "initial",

    "heartbeat.interval.ms": "15000",
    "heartbeat.action.query": "INSERT INTO public.debezium_heartbeat (ts) VALUES (now()) ON CONFLICT DO NOTHING",

    "decimal.handling.mode": "double",
    "time.precision.mode": "connect",

    "tombstones.on.delete": "true",

    "errors.tolerance": "none",
    "errors.deadletterqueue.topic.name": "debezium.dlq.orders",
    "errors.deadletterqueue.context.headers.enable": "true",

    "transforms": "route,unwrap",
    "transforms.route.type": "org.apache.kafka.connect.transforms.ReplaceField$Value",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "transforms.unwrap.drop.tombstones": "false",
    "transforms.unwrap.delete.handling.mode": "rewrite",
    "transforms.unwrap.add.fields": "op,ts_ms,source.ts_ms"
  }
}

Let’s break down the non-obvious parts.

heartbeat.interval.ms + heartbeat.action.query: This is critical. If you have tables that change infrequently, Debezium never gets a chance to advance the LSN (log sequence number) stored in its offsets. The replication slot stalls at an old position, Postgres can’t clean the WAL, and you get WAL bloat even though your connector is "healthy". The heartbeat inserts a dummy row into a tiny table you create, giving Debezium a constant stream of WAL events to acknowledge. Create the heartbeat table:

CREATE TABLE public.debezium_heartbeat (
  id SERIAL PRIMARY KEY,
  ts TIMESTAMPTZ DEFAULT NOW()
);
GRANT INSERT, UPDATE ON public.debezium_heartbeat TO debezium;

slot.drop.on.stop: false: Default is false, but say it explicitly. If this were true, stopping the connector would drop the slot, destroying the resume position. Restarting would trigger a full snapshot.

ExtractNewRecordState SMT: Debezium’s native event format is verbose — it wraps before, after, source, op, and ts_ms in a single envelope. Most downstream consumers don’t want the envelope; they want the row. ExtractNewRecordState unwraps it to just the after payload. Adding op and ts_ms back via add.fields gives you the operation type and timing without the full envelope.

Snapshot Modes — Choose Carefully

The snapshot.mode decision has consequences you’ll feel much later:

  • initial — full table scan on first start, then streaming. The right choice when you need historical data in the topic.
  • initial_only — takes a snapshot and stops. Useful for one-time migrations.
  • schema_only — captures schema only, then streams from current LSN. Use this when you don’t want history in Kafka but need the connector to know the schema for new events.
  • never — doesn’t snapshot at all. Assumes the topic already has data or you don’t care about history. Dangerous if the connector offset gets lost.
  • always — full snapshot on every restart. Almost never what you want in production.

Gotcha: initial on a multi-hundred-million-row table will stress your Postgres for hours and produce enormous Kafka topics. Plan the snapshot window, possibly set snapshot.max.threads and snapshot.fetch.size to throttle it, and make sure your Kafka topic retention is set appropriately before the snapshot floods it.

MySQL / MariaDB Connector

The MySQL connector has a different risk profile. Binlog position, not a slot, is the anchor. If the binlog rotates before the connector catches up, you get Could not find first log file name in binary log index file and you’re forced to re-snapshot.

{
  "name": "mysql-users-connector",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "mysql",
    "database.port": "3306",
    "database.user": "debezium",
    "database.password": "strong-password-here",
    "database.server.id": "184054",
    "topic.prefix": "mysql",
    "database.include.list": "users_db",
    "table.include.list": "users_db.users,users_db.sessions",

    "snapshot.mode": "initial",

    "include.schema.changes": "true",
    "database.history.kafka.topic": "schema-changes.mysql-users",
    "database.history.kafka.bootstrap.servers": "kafka:29092",

    "snapshot.locking.mode": "minimal",

    "binary.handling.mode": "base64",
    "decimal.handling.mode": "string",

    "heartbeat.interval.ms": "10000",

    "errors.tolerance": "none",
    "errors.deadletterqueue.topic.name": "debezium.dlq.mysql-users",
    "errors.deadletterqueue.context.headers.enable": "true"
  }
}

database.history.kafka.topic is where Debezium stores the schema history. It’s not a changelog — it’s an internal log Debezium uses to reconstruct the schema at any point in time. Never delete this topic or compact it. Set cleanup.policy=delete with a very long or infinite retention.

snapshot.locking.mode: minimal takes a global read lock only during the brief moment it captures the binlog position, then releases it. Default minimal is almost always right. none risks a slightly inconsistent snapshot but is sometimes necessary on read-heavy systems.

Gotcha: server.id must be unique across all MySQL slaves, including other Debezium connectors against the same cluster. Conflicts cause bizarre replication errors that are annoying to diagnose.

Schema Evolution

This is where most setups quietly rot. Columns get added, types change, tables get renamed. What happens to your pipeline?

With Avro and Schema Registry, Debezium will register and evolve schemas automatically, within the bounds of compatibility rules. The default compatibility in Confluent Schema Registry is BACKWARD — new schema can read data written with the old schema. This means:

  • Adding an optional column: compatible, safe.
  • Adding a required column without a default: breaks backward compatibility, Schema Registry will reject the new schema registration and your connector will go into an error state.
  • Changing a column type (e.g., INT to BIGINT): not compatible by default.

Set compatibility to FORWARD or FULL depending on your consumer contracts, and make sure your DDL changes go through a migration that Debezium can tolerate. Adding columns with defaults is the safest path.

If you use include.schema.changes: true (MySQL) or just rely on pgoutput publications (Postgres), Debezium will track schema changes. But the downstream topic consumers are your responsibility.

Dead Letter Queues and Error Handling

"errors.tolerance": "none" in the config above is intentional. For CDC, failing loudly is almost always better than skipping corrupt events silently. But you still need a DLQ for the cases where a single bad record would stall the entire connector.

A reasonable production policy:

  • Set errors.tolerance: all only if you have a consumer reading the DLQ and alerting on it.
  • Set errors.deadletterqueue.topic.name to a monitored topic.
  • Set errors.deadletterqueue.context.headers.enable: true — the headers carry the exception, the original topic/partition/offset, and the connector name. Without them, debugging DLQ messages is painful.

Build a simple DLQ inspector. The headers tell you everything:

# Peek at DLQ message headers with kafkacat
kafkacat -b localhost:9092 \
  -t debezium.dlq.orders \
  -C -f 'Headers: %h\nValue: %s\n---\n' \
  -o beginning -e

Monitoring

Debezium exposes JMX metrics. In production, scrape them with the JMX Exporter for Prometheus.

The metrics that matter most:

  • debezium_postgres_connector_metrics_millisecondsbehindoperationalstream — lag in milliseconds. Alert at 60s for OLTP, 300s for batch workloads.
  • debezium_postgres_connector_metrics_totalnumberofeventsfiltered — if this suddenly spikes, a filter or SMT is dropping events you might care about.
  • debezium_postgres_connector_metrics_lasttransactionid — lets you correlate with pg_replication_slots.confirmed_flush_lsn.
  • kafka_connect_connector_task_metrics_running_ratio — should be 1.0. Drop means task restarts.

Check the replication slot directly from Postgres too:

SELECT slot_name,
       pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn)) AS lag_size,
       active,
       wal_status
FROM pg_replication_slots
WHERE slot_name = 'debezium_orders';

wal_status going to lost means the slot lost its WAL position — you’ve already missed events and need a re-snapshot.

Gotchas Condensed

WAL bloat: Already covered, but worth repeating. Set an alert on pg_replication_slots lag size. If it crosses 5GB, something is wrong. If it crosses 20GB, you’re in a disk emergency.

Offset topic compaction: Debezium stores connector offsets in debezium-offsets. This topic must have cleanup.policy=compact. If it somehow gets deleted or has delete policy, the connector loses its position on restart and re-snapshots. Verify this on first setup.

Table renames and drops: Debezium doesn’t handle table drops gracefully on Postgres — the connector will error. Always update the connector’s table.include.list before dropping a tracked table, then restart the connector cleanly.

tombstones.on.delete: true: When a row is deleted, Debezium emits a delete event followed by a tombstone (null value, same key). The tombstone triggers Kafka log compaction to remove the key’s history. Disable tombstones only if you’re using a sink that can’t handle null values and you’ve accepted that deleted rows will persist in compacted topics forever.

Connector task vs. worker restart: Restarting the Kafka Connect worker is not the same as restarting a connector task. A task restart resumes from the committed offset. A full worker restart under load can cause duplicate delivery during the offset catch-up window. Build idempotent sinks.

Snapshot isolation level: Debezium’s snapshot runs in REPEATABLE READ for Postgres. If you have long-running transactions in flight during the snapshot, you may miss concurrent writes. The pgoutput streaming that follows the snapshot will catch them, but there’s a brief window where change ordering matters. Don’t run snapshots on read replicas for this reason — use the primary.

Production-Ready Connector Management Script

Wrap your connector lifecycle in a script rather than raw curl:

#!/usr/bin/env bash
# debezium-ctl.sh — lightweight connector management

set -euo pipefail

CONNECT_URL="${CONNECT_URL:-https://cd-linux.club:8083}"
CONNECTOR_DIR="${1:-./connectors}"

cmd="${2:-status}"

list_connectors() {
  curl -s "$CONNECT_URL/connectors" | jq -r '.[]'
}

status_all() {
  for c in $(list_connectors); do
    state=$(curl -s "$CONNECT_URL/connectors/$c/status" | jq -r '.connector.state')
    tasks=$(curl -s "$CONNECT_URL/connectors/$c/status" | jq -r '[.tasks[].state] | join(",")')
    echo "$c  connector=$state  tasks=$tasks"
  done
}

apply_connector() {
  local file="$1"
  local name
  name=$(jq -r '.name' "$file")
  if curl -s "$CONNECT_URL/connectors/$name" | jq -e '.name' > /dev/null 2>&1; then
    echo "Updating $name"
    curl -s -X PUT "$CONNECT_URL/connectors/$name/config" \
      -H "Content-Type: application/json" \
      -d "$(jq '.config' "$file")" | jq .
  else
    echo "Creating $name"
    curl -s -X POST "$CONNECT_URL/connectors" \
      -H "Content-Type: application/json" \
      -d @"$file" | jq .
  fi
}

restart_failed() {
  for c in $(list_connectors); do
    curl -s "$CONNECT_URL/connectors/$c/status" | jq -r '.tasks[] | select(.state=="FAILED") | .id' | while read -r tid; do
      echo "Restarting failed task $tid of $c"
      curl -s -X POST "$CONNECT_URL/connectors/$c/tasks/$tid/restart"
    done
  done
}

case "$cmd" in
  status) status_all ;;
  apply)  apply_connector "$CONNECTOR_DIR" ;;
  restart-failed) restart_failed ;;
  *) echo "Usage: $0 [connector.json] [status|apply|restart-failed]" && exit 1 ;;
esac

Keep connector JSON definitions in version control alongside this script. apply is idempotent — it creates or updates. Run it from CI on every merge to the connector config directory.

A Note on Exactly-Once

Debezium delivers at-least-once semantics. Events can be duplicated on connector restart. Kafka Transactions can give you exactly-once within the Kafka ecosystem (Debezium 2.x supports this with Kafka Connect’s exactly-once support when configured), but the moment data leaves Kafka into a sink — a database, an API, an S3 file — you’re back to at-least-once unless the sink is idempotent.

Design your sinks to be idempotent. Use upserts keyed on primary key + ts_ms. Use outbox pattern tables for event publication if exactly-once delivery to external systems is a hard requirement — Debezium has first-class outbox support via the EventRouter SMT.

What to Actually Run in Production

The setup above works. For a real production deployment, a few additional concerns:

Run Kafka Connect as a distributed cluster — at least two workers for failover. A single Kafka Connect worker is a SPOF. Debezium tasks are just Kafka Connect tasks and can be rebalanced across workers automatically.

Use a secrets manager. Passwords in plain JSON connector configs stored in Kafka topics is not great. Kafka Connect supports external secret providers for AWS Secrets Manager, Vault, and others. Wire them up before you promote to production.

Enable SSL/TLS between Debezium and your databases. The connector configs support database.ssl.mode for Postgres and MySQL. On managed databases (RDS, Cloud SQL), TLS is often required anyway.

Consider topic naming conventions early. Debezium defaults to {prefix}.{schema}.{table}. Once consumers are built against those topic names, renaming is painful. Decide on your naming scheme before you go live.

The gap between "connector running" and "production-grade pipeline" is real, but it’s not mysterious. It’s WAL slots, schema evolution, offset durability, and idempotent sinks. Handle those four things and Debezium will run for months without waking you up.

Leave a comment

👁 Views: 10,370 · Unique visitors: 14,506