CDC Change Data Capture

Change Data Capture (CDC) is a pattern for efficiently detecting and propagating data changes—inserts, updates, and deletes—from a source database to downstream systems. It matters when real-time or near-real-time data synchronization is required across services, analytics pipelines, caches, or event-driven architectures. CDC is the foundation for building event-sourced systems, data warehouses, and reactive microservices. It replaces inefficient polling, full-table scans, and manual ETL triggers with an automated, incremental, and scalable way to move data as it evolves.

Detecting Changes

Change Data Capture begins with identifying *what* changed and *when*. The source system must record modifications to its data over time. The most common approach is to use a log-based mechanism that captures changes as events in a structured format. The source database writes a log of changes to a persistent, append-only stream—typically a transaction log or WAL (Write-Ahead Log)—and downstream systems read this log to discover new or modified data.

The log must include:

• The table and row identifier (e.g., primary key value)
• The operation type: INSERT, UPDATE, DELETE
• The timestamp of when the change was committed
• The new and old values of the changed fields

The simplest form of CDC uses a change table in the source database: a table that records each change as a row. A trigger or a background process writes to this table after every DML operation. The change table can be queried by a downstream service to fetch all changes since a given timestamp or sequence number.

Example: Change Table Schema

CREATE TABLE change_log (
    id BIGSERIAL PRIMARY KEY,
    table_name VARCHAR(64) NOT NULL,
    record_id VARCHAR(64) NOT NULL,
    operation_type VARCHAR(10) NOT NULL CHECK (operation_type IN ('INSERT', 'UPDATE', 'DELETE')),
    old_values JSONB,
    new_values JSONB,
    changed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    processed BOOLEAN DEFAULT FALSE
);

A service reading this table uses the changed_at or id as a cursor to fetch changes incrementally. A typical read loop looks like:

SELECT id, table_name, record_id, operation_type, old_values, new_values, changed_at
FROM change_log
WHERE processed = FALSE
  AND changed_at > '2024-04-05T10:00:00Z'
ORDER BY changed_at
LIMIT 1000;

When a batch is processed, the service marks its entries as processed = TRUE using a UPDATE or UPSERT to avoid reprocessing.

Streaming Changes

For higher throughput and lower latency, CDC systems stream changes directly from the database’s native transaction log. This approach eliminates the need for triggers and reduces the overhead of writing to a change table.

The key requirement is a log reader that can consume the log sequentially, starting from a known offset or LSN (Log Sequence Number). The reader maintains a cursor (a checkpoint) that tracks the last processed log entry.

Example: Log Reader in Python

import psycopg2
from psycopg2.extras import RealDictCursor

def read_changes_from_wal(connection, start_lsn: str):
    cursor = connection.cursor(cursor_factory=RealDictCursor)
    cursor.execute(f"""
        SELECT
            lsn,
            xid,
            op,
            relname,
            old_values,
            new_values,
            ts
        FROM pg_wal_scan
        WHERE lsn >= %s
        ORDER BY lsn
    """, [start_lsn])

    while True:
        batch = cursor.fetchmany(1000)
        if not batch:
            break

        for record in batch:
            yield {
                "table": record["relname"],
                "key": record["old_values"]["id"] if record["op"] == "DELETE" else record["new_values"]["id"],
                "op": record["op"],
                "old": record["old_values"],
                "new": record["new_values"],
                "timestamp": record["ts"],
                "lsn": record["lsn"]
            }

        # Update checkpoint
        last_lsn = batch[-1]["lsn"]
        cursor.execute(
            "INSERT INTO cdc_checkpoints (name, lsn) VALUES (%s, %s) "
            "ON CONFLICT (name) DO UPDATE SET lsn = EXCLUDED.lsn",
            ["main", last_lsn]
        )

        connection.commit()

This reader reads from PostgreSQL’s pg_wal_scan view or uses pg_logical_slot_advance and pg_logical_slot_get_changes functions. The lsn serves as a durable, incrementing cursor that survives restarts and can be used to resume from any point.

Applying Changes to a Target System

Once changes are detected, they must be applied to a target system—another database, a message queue, a data warehouse, or a cache. The target system must support idempotent, incremental updates.

Example: Applying Changes to a Search Index

import requests
from concurrent.futures import ThreadPoolExecutor

def apply_change(change):
    url = f"http://search-service:9200/index/_doc/{change['table']}/{change['key']}"
    data = change["new"] if change["op"] == "INSERT" or change["op"] == "UPDATE" else None

    method = "POST" if change["op"] == "INSERT" else "PUT"
    headers = {"Content-Type": "application/json"}

    try:
        response = requests.request(
            method=method,
            url=url,
            json=data,
            headers=headers,
            timeout=10
        )
        response.raise_for_status()
        return {"success": True, "status": response.status_code}
    except requests.RequestException as e:
        return {"success": False, "error": str(e)}

The target system must handle:

• Idempotency: multiple applications of the same change do not alter the result beyond the first.
• Conflict resolution: when multiple changes arrive for the same record, which version wins?
• Schema evolution: the structure of new_values or old_values may change over time, requiring backward compatibility.

Idempotency via Versioned Keys

To ensure idempotency, the target system can use a versioned key derived from the record's primary key and the change’s timestamp or LSN:

{
  "table": "users",
  "key": "12345",
  "version": "2024-04-05T10:01:23.456Z",
  "operation": "UPDATE",
  "data": { "name": "Alice", "email": "[email protected]" }
}

When applying a change, the target system checks if a record with the same table, key, and version exists. If it does, the change is a no-op. If not, it creates or updates the record.

Handling Failures and Reconciliation

CDC systems fail silently. The most common issues are:

• Log reader crashes and loses its cursor
• Target system is unavailable during change application
• Duplicate or out-of-order change events
• Schema drift in the source data

Checkpointing and Recovery

A CDC system must checkpoint the log reader’s progress. The checkpoint is a durable, persistent record of the last processed entry. On restart, the reader reads from the last checkpoint.

A typical checkpoint table:

CREATE TABLE cdc_checkpoints (
    name VARCHAR(64) PRIMARY KEY,
    lsn TEXT NOT NULL,         -- e.g., '0/1A2B3C4D'
    last_seen_at TIMESTAMPTZ NOT NULL,
    processed_count BIGINT DEFAULT 0
);

On startup, the reader:

1. Fetches the latest checkpoint for its name (e.g., main, orders).
2. Initializes the log reader from the lsn.
3. Processes changes.
4. Updates the checkpoint after every batch.

If the reader crashes after processing 500 changes but before updating the checkpoint, the next run resumes from the same LSN, avoiding duplicates.

Duplicate Processing

When the target system receives the same change twice—due to network failure, idempotency mismatch, or retry logic—the system must deduplicate.

The solution is deduplication via a deduplication ID. The CDC system generates a unique, stable identifier for each change. This ID is derived from:

• The table name
• The record’s primary key
• The operation type
• The timestamp or LSN of the change

CREATE TABLE change_dedup (
    dedup_id VARCHAR(255) PRIMARY KEY,
    change_id BIGINT,
    processed_at TIMESTAMPTZ NOT NULL,
    processed_by VARCHAR(64)
);

Before applying a change, the target system checks if dedup_id exists in change_dedup. If it does, the change is a duplicate and is skipped. Otherwise, the change is applied and dedup_id is recorded.

Schema Drift and Evolution

As the source database evolves—adding columns, changing types, or reorganizing tables—the structure of new_values and old_values may change. The CDC reader must handle schema drift.

The reader can emit a schema snapshot at regular intervals or on every schema change. This snapshot includes:

• The table name and version
• The current schema of the table (column names, types, constraints)
• A mapping from field names to their positions or IDs

The target system uses this schema to deserialize new_values and old_values correctly. A change event with a schema version mismatch triggers a migration or warning.

Configuration and Tuning

CDC systems are tuned by adjusting the following parameters:

• Batch size: how many changes to read and process per iteration. Smaller batches reduce latency; larger batches improve throughput.
• Checkpoint interval: how often to persist the log reader’s cursor. Frequent checkpoints reduce recovery time but increase I/O.
• Backoff strategy: when the target system fails, how to retry changes. Exponential backoff with jitter is common.
• Change filtering: only process changes for certain tables, or filter by operation type.

Example: Configuration File (YAML)

cdc:
  source:
    type: postgres
    connection_string: "host=db.example.com user=cdc_user dbname=app"
    log_reader:
      start_lsn: "0/1A2B3C4D"
      batch_size: 1000
      checkpoint_interval: 300  # seconds
      backoff:
        initial_delay: 1000
        max_delay: 30000
        max_retries: 10
      filters:
        tables:
          - users
          - orders
        operations:
          - INSERT
          - UPDATE
        max_age: "1h"
  target:
    type: http
    endpoint: "http://search-service:9200/index/_doc"
    max_retries: 5
    timeout: 10
    deduplication:
      enabled: true
      dedup_id_template: "{{table}}/{{key}}/{{operation}}/{{lsn}}"

Common Pitfalls

• Log reader starts from wrong LSN: a reader that begins at 0/00000000 but the database has already been active for hours may miss changes.
• Checkpointing on commit, not on success: updating the checkpoint before the change is fully applied leads to data loss on failure.
• Assuming order is guaranteed: in distributed systems, changes may arrive out of order. A simple ORDER BY lsn is not enough; systems must validate or re-order changes.
• Ignoring the old_values: a common mistake is only applying new_values and ignoring old_values, leading to incomplete state tracking.
• Not handling DELETE operations correctly: a DELETE change must be processed even if the target record doesn’t exist; it must be removed.

Monitoring and Observability

A CDC pipeline should emit logs, metrics, and alerts. Key observability signals:

• Change rate: events per second
• Latency: time from change in source to delivery in target
• Checkpoint lag: difference between latest log entry and last checkpoint
• Error rate: number of failed change applications
• Duplicate rate: percentage of changes that are duplicates

A health check endpoint returns:

{
  "status": "healthy",
  "source": {
    "last_lsn": "0/1A2B3C4D",
    "checkpoint_lsn": "0/1A2B3C4D",
    "lag_seconds": 12.5
  },
  "target": {
    "last_applied": "2024-04-05T10:01:30Z",
    "last_checkpointed": "2024-04-05T10:00:00Z",
    "retries": 34,
    "duplicates": 12
  },
  "errors": [
    {
      "timestamp": "2024-04-05T10:00:15Z",
      "change_id": 12345,
      "error": "HTTP 503: Service Unavailable"
    }
  ]
}

This health check is polled by a service mesh, a monitoring system, or a dashboard. Alerts trigger when lag exceeds 30 seconds or when error rate exceeds 1%.

This page was rewritten on 18 October 2026. It replaced a templated version that was not specific to its own title. How these pages are written and checked. If anything here is wrong, tell us at [email protected] and we will correct it.