Change Data Capture With Debezium — Stream Database Changes to Anywhere in 2026

Introduction

Why This Matters

Every time a row changes in your database, downstream systems need to know: search indexes need re-indexing, caches need invalidation, analytics pipelines need the new data, and event-driven microservices need the change event. The naive solution — dual writes from the application layer — is fragile, because the application can fail between the database write and the Kafka publish, leaving systems out of sync.

Change Data Capture solves this by reading changes directly from the database transaction log. The log is the authoritative record: if the transaction committed, the change event will be emitted. No dual-write race conditions.

How Debezium CDC Works

Debezium runs as a Kafka Connect connector and reads directly from the database replication stream:

PostgreSQL WAL (Write-Ahead Log)
         |
         v
Debezium PostgreSQL Connector
         |
         v
Kafka Connect Worker
         |
         v
Kafka Topic: mydb.public.users
         |
    _____|_____
   |           |
   v           v
Elasticsearch  Redis Cache
(search index) (invalidation)

For PostgreSQL, Debezium uses logical replication. For MySQL, it reads the binlog. The database must be configured to enable this before Debezium can connect.

PostgreSQL Configuration for CDC

-- postgresql.conf changes required
-- wal_level = logical (enables logical replication slots)
-- max_wal_senders = 4 (concurrent replication connections)
-- max_replication_slots = 4 (persistent replication slots)
 
-- Create a replication user
CREATE USER debezium_user WITH REPLICATION LOGIN PASSWORD 'secure_password';
GRANT SELECT ON ALL TABLES IN SCHEMA public TO debezium_user;
GRANT USAGE ON SCHEMA public TO debezium_user;
 
-- Enable REPLICA IDENTITY FULL for tables that need old values on UPDATE/DELETE
-- Default REPLICA IDENTITY only includes primary key in change events
ALTER TABLE users REPLICA IDENTITY FULL;
ALTER TABLE orders REPLICA IDENTITY FULL;
 
-- Check current WAL level (must be 'logical')
SHOW wal_level;

Debezium Connector Configuration

{
  "name": "postgres-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres.internal",
    "database.port": "5432",
    "database.user": "debezium_user",
    "database.password": "${file:/kafka/custom-config/secrets.properties:postgres.password}",
    "database.dbname": "mydb",
    "database.server.name": "mydb",
    "slot.name": "debezium_slot",
    "plugin.name": "pgoutput",
    "table.include.list": "public.users,public.orders,public.products",
    "heartbeat.interval.ms": "10000",
    "snapshot.mode": "initial",
    "publication.name": "debezium_publication",
    "decimal.handling.mode": "double",
    "time.precision.mode": "connect",
    "tombstones.on.delete": "true",
    "topic.prefix": "mydb",
    "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"
  }
}

Processing CDC Events in Node.js

import { Kafka } from 'kafkajs';
 
interface DebeziumEvent {
  before: Record<string, unknown> | null;
  after: Record<string, unknown> | null;
  op: 'c' | 'u' | 'd' | 'r'; // create, update, delete, read (snapshot)
  ts_ms: number;
  table: string;
}
 
const kafka = new Kafka({
  clientId: 'cdc-consumer',
  brokers: ['kafka:9092'],
});
 
const consumer = kafka.consumer({ groupId: 'cdc-processor' });
 
await consumer.connect();
await consumer.subscribe({
  topics: ['mydb.public.users', 'mydb.public.orders'],
  fromBeginning: false,
});
 
await consumer.run({
  eachMessage: async ({ topic, message }) => {
    if (!message.value) return; // tombstone (delete marker)
 
    const event: DebeziumEvent = JSON.parse(message.value.toString());
    const table = topic.split('.').pop()!;
 
    switch (event.op) {
      case 'c':
        await handleInsert(table, event.after!);
        break;
      case 'u':
        await handleUpdate(table, event.before, event.after!);
        break;
      case 'd':
        await handleDelete(table, event.before!);
        break;
      case 'r':
        await handleSnapshot(table, event.after!);
        break;
    }
  },
});
 
async function handleUpdate(
  table: string,
  before: Record<string, unknown> | null,
  after: Record<string, unknown>
): Promise<void> {
  if (table === 'users') {
    const userId = after.id as number;
 
    // Invalidate cache
    await redis.del(`user:${userId}`);
 
    // Update search index only if searchable fields changed
    if (before?.name !== after.name || before?.email !== after.email) {
      await elasticsearch.update({
        index: 'users',
        id: String(userId),
        doc: { name: after.name, email: after.email },
      });
    }
  }
}

Handling Schema Evolution

When you add a column to a table, Debezium events automatically include the new field. Use a schema registry to manage compatibility:

import { SchemaRegistry } from '@kafkajs/confluent-schema-registry';
 
const registry = new SchemaRegistry({ host: 'http://schema-registry:8081' });
 
// Register schema with backward compatibility
const schemaId = await registry.register({
  type: SchemaType.AVRO,
  schema: JSON.stringify({
    type: 'record',
    name: 'User',
    namespace: 'mydb.public',
    fields: [
      { name: 'id', type: 'long' },
      { name: 'email', type: 'string' },
      { name: 'name', type: 'string' },
      // New optional field — backward compatible
      { name: 'phone', type: ['null', 'string'], default: null },
    ],
  }),
}, { subject: 'mydb.public.users-value' });

Monitoring CDC Lag

Replication lag is the most important CDC health metric:

-- Check replication slot lag in PostgreSQL
SELECT
  slot_name,
  plugin,
  slot_type,
  database,
  active,
  pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn)) AS lag_size,
  confirmed_flush_lsn
FROM pg_replication_slots
WHERE slot_name = 'debezium_slot';

// Prometheus metrics for CDC monitoring
import { Gauge } from 'prom-client';
 
const cdcLagGauge = new Gauge({
  name: 'cdc_replication_lag_bytes',
  help: 'Bytes of WAL not yet consumed by Debezium',
  labelNames: ['slot_name'],
});
 
// Alert if lag exceeds 100MB (signal of processing bottleneck)

Common Mistakes

Not setting REPLICA IDENTITY FULL. Without it, DELETE events only include the primary key — no other column values. UPDATE events only include the primary key in before. Set REPLICA IDENTITY FULL on tables where you need the full old row.

Leaving inactive replication slots. PostgreSQL holds WAL files for every active replication slot. An abandoned slot (Debezium stopped but not deleted) causes unbounded WAL accumulation, eventually filling the disk.

Processing events without idempotency. CDC guarantees at-least-once delivery. Your consumers must handle duplicate events gracefully using the lsn (log sequence number) as an idempotency key.

No schema registry. Without a schema registry, a column rename or type change in the database can break all consumers simultaneously.

Snapshot mode confusion. snapshot.mode: initial replays the entire table on first start. For large tables this can take hours. Use schema_only if you only need changes going forward.

Best Practices

Enable REPLICA IDENTITY FULL on all tables you want full before/after change events for
Use the ExtractNewRecordState SMT (Single Message Transform) to simplify event payloads
Monitor replication slot WAL lag and alert at 50MB — an inactive slot can fill your disk
Use a schema registry with backward compatibility to safely evolve event schemas
Include the LSN as an idempotency key in all CDC consumer processing logic
Delete inactive replication slots immediately to prevent WAL accumulation

Key Takeaways

Debezium reads PostgreSQL WAL logs and MySQL binlogs directly, guaranteeing that any committed transaction produces a change event — eliminating dual-write race conditions.
PostgreSQL requires wal_level = logical and a dedicated replication user with the REPLICATION privilege before Debezium can connect.
REPLICA IDENTITY FULL on a table includes all column values in UPDATE and DELETE events, not just the primary key.
Inactive replication slots cause PostgreSQL to retain WAL files indefinitely — always drop slots when a Debezium connector is permanently removed.
CDC events are delivered at least once; consumer logic must be idempotent using the LSN (log sequence number) as a deduplication key.
A schema registry with backward compatibility allows adding optional fields to CDC event schemas without breaking existing consumers.
Replication lag in bytes (measured from pg_replication_slots) is the primary health metric for a CDC pipeline — alert at 50–100MB.
snapshot.mode: schema_only skips replaying existing rows and only captures changes going forward, making it suitable for large tables.

Introduction#

Why This Matters#

How Debezium CDC Works#

PostgreSQL Configuration for CDC#

Debezium Connector Configuration#

Processing CDC Events in Node.js#

Handling Schema Evolution#

Monitoring CDC Lag#

Common Mistakes#

Best Practices#

Key Takeaways#

Related reading