Implementing Database Change Data Capture Pipelines
Build CDC pipelines that stream database changes downstream in real time: log-based capture with Debezium, event formats, schema evolution, exactly-once.
Polling a database every few seconds to detect changes wastes resources and introduces latency. Change Data Capture (CDC) reads the database's own transaction log—the write-ahead log (WAL) in PostgreSQL, the binlog in MySQL—and streams every insert, update, and delete as an event. This gives you real-time data replication without impacting the source database's performance.
CDC is the backbone of modern data architectures: it powers search index updates, cache invalidation, analytics pipelines, and cross-service data synchronization without coupling producers to consumers.
Log-Based CDC with Debezium
Debezium reads the database transaction log and publishes change events to Kafka. The source database doesn't need any modification—no triggers, no polling queries, no application-level event publishing.
# ❌ Polling-based approach — high latency, wasteful queries
# SELECT * FROM orders
# WHERE updated_at > :last_check_time
# Run every 5 seconds across all tables
# Misses deletes, creates load on source database// ✅ Debezium connector configuration for PostgreSQL CDC
interface DebeziumConnectorConfig {
name: string;
config: {
"connector.class": string;
"database.hostname": string;
"database.port": number;
"database.user": string;
"database.dbname": string;
"database.server.name": string;
"plugin.name": string;
"slot.name": string;
"publication.name": string;
"table.include.list": string;
"transforms": string;
"transforms.unwrap.type": string;
"transforms.unwrap.drop.tombstones": string;
"key.converter": string;
"value.converter": string;
"value.converter.schemas.enable": string;
};
}
const ordersCdcConnector: DebeziumConnectorConfig = {
name: "orders-cdc-connector",
config: {
"connector.class":
"io.debezium.connector.postgresql.PostgresConnector",
"database.hostname": "db.internal",
"database.port": 5432,
"database.user": "cdc_reader",
"database.dbname": "production",
"database.server.name": "prod-orders",
"plugin.name": "pgoutput",
"slot.name": "orders_slot",
"publication.name": "orders_publication",
"table.include.list": "public.orders,public.order_items",
"transforms": "unwrap",
"transforms.unwrap.type":
"io.debezium.transforms.ExtractNewRecordState",
"transforms.unwrap.drop.tombstones": "false",
"key.converter":
"org.apache.kafka.connect.json.JsonConverter",
"value.converter":
"org.apache.kafka.connect.json.JsonConverter",
"value.converter.schemas.enable": "false",
},
};The pgoutput plugin reads PostgreSQL's logical replication stream. The ExtractNewRecordState transform flattens Debezium's envelope format into the after-state of each row, simplifying downstream processing.
Processing Change Events
Each CDC event contains the operation type, the changed data, and metadata about the source. Here's how to consume and route these events.
interface CdcEvent {
op: "c" | "u" | "d" | "r"; // create, update, delete, read (snapshot)
before: Record<string, unknown> | null;
after: Record<string, unknown> | null;
source: {
version: string;
connector: string;
name: string;
ts_ms: number;
db: string;
schema: string;
table: string;
lsn: number;
};
ts_ms: number;
}
type EventHandler = (event: CdcEvent) => Promise<void>;
class CdcEventRouter {
private handlers: Map<string, EventHandler[]> = new Map();
on(table: string, handler: EventHandler): void {
const existing = this.handlers.get(table) ?? [];
existing.push(handler);
this.handlers.set(table, existing);
}
async process(event: CdcEvent): Promise<void> {
const table = event.source.table;
const handlers = this.handlers.get(table) ?? [];
for (const handler of handlers) {
try {
await handler(event);
} catch (error) {
console.error(
`Handler failed for ${table}:`,
error
);
// Dead letter queue for failed events
await this.sendToDeadLetter(event, error as Error);
}
}
}
private async sendToDeadLetter(
event: CdcEvent,
error: Error
): Promise<void> {
console.error(
JSON.stringify({
type: "dead_letter",
table: event.source.table,
operation: event.op,
lsn: event.source.lsn,
error: error.message,
timestamp: new Date().toISOString(),
})
);
}
}
// Wire up handlers for different tables
const router = new CdcEventRouter();
router.on("orders", async (event) => {
if (event.op === "c" || event.op === "u") {
await updateSearchIndex("orders", event.after);
await invalidateCache(`order:${event.after?.id}`);
}
if (event.op === "d") {
await removeFromSearchIndex("orders", event.before?.id);
await invalidateCache(`order:${event.before?.id}`);
}
});
router.on("order_items", async (event) => {
if (event.op === "c") {
await updateAnalytics("new_item", event.after);
}
});Handling Schema Evolution
Database schemas change. Columns get added, renamed, or dropped. Your CDC pipeline must handle these changes without breaking downstream consumers.
interface SchemaVersion {
version: number;
table: string;
columns: Map<string, ColumnDef>;
migrations: SchemaMigration[];
}
interface ColumnDef {
name: string;
type: string;
nullable: boolean;
defaultValue?: unknown;
}
interface SchemaMigration {
fromVersion: number;
toVersion: number;
transform: (record: Record<string, unknown>) => Record<string, unknown>;
}
class SchemaRegistry {
private schemas: Map<string, SchemaVersion[]> = new Map();
register(schema: SchemaVersion): void {
const existing = this.schemas.get(schema.table) ?? [];
existing.push(schema);
existing.sort((a, b) => a.version - b.version);
this.schemas.set(schema.table, existing);
}
evolve(
table: string,
record: Record<string, unknown>,
fromVersion: number,
toVersion: number
): Record<string, unknown> {
const versions = this.schemas.get(table);
if (!versions) return record;
let current = { ...record };
for (const schema of versions) {
for (const migration of schema.migrations) {
if (
migration.fromVersion >= fromVersion &&
migration.toVersion <= toVersion
) {
current = migration.transform(current);
}
}
}
return current;
}
}
// Example: handling a column rename
const registry = new SchemaRegistry();
registry.register({
version: 2,
table: "orders",
columns: new Map([
["id", { name: "id", type: "uuid", nullable: false }],
[
"customer_email",
{ name: "customer_email", type: "text", nullable: false },
],
]),
migrations: [
{
fromVersion: 1,
toVersion: 2,
transform: (record) => {
// Column renamed: email → customer_email
const { email, ...rest } = record;
return { ...rest, customer_email: email };
},
},
],
});Exactly-Once Delivery Semantics
CDC events must be processed exactly once. Duplicate processing leads to incorrect counts, duplicate notifications, and data inconsistency.
interface ProcessedEvent {
eventId: string;
lsn: number;
processedAt: Date;
}
class IdempotentProcessor {
private processedEvents: Map<string, ProcessedEvent> = new Map();
private generateEventId(event: CdcEvent): string {
// Unique ID from source position + table + operation
return `${event.source.name}:${event.source.lsn}:${event.source.table}:${event.op}`;
}
async processOnce(
event: CdcEvent,
handler: (event: CdcEvent) => Promise<void>
): Promise<{ processed: boolean; reason?: string }> {
const eventId = this.generateEventId(event);
// Check if already processed
if (this.processedEvents.has(eventId)) {
return {
processed: false,
reason: "duplicate",
};
}
try {
// Process within a transaction that also records
// the event as processed
await handler(event);
this.processedEvents.set(eventId, {
eventId,
lsn: event.source.lsn,
processedAt: new Date(),
});
return { processed: true };
} catch (error) {
// Don't mark as processed — allow retry
throw error;
}
}
getLastProcessedLsn(): number {
let maxLsn = 0;
for (const event of this.processedEvents.values()) {
if (event.lsn > maxLsn) maxLsn = event.lsn;
}
return maxLsn;
}
}Monitoring CDC Pipeline Health
A CDC pipeline that silently falls behind is worse than no CDC at all—downstream systems serve stale data without knowing it.
interface CdcPipelineMetrics {
replicationLag: number; // milliseconds
eventsPerSecond: number;
lastProcessedLsn: number;
lastEventTimestamp: Date;
errorCount: number;
deadLetterCount: number;
}
class CdcMonitor {
private metrics: CdcPipelineMetrics = {
replicationLag: 0,
eventsPerSecond: 0,
lastProcessedLsn: 0,
lastEventTimestamp: new Date(),
errorCount: 0,
deadLetterCount: 0,
};
private eventTimestamps: number[] = [];
recordEvent(event: CdcEvent): void {
const now = Date.now();
const eventTime = event.source.ts_ms;
this.metrics.replicationLag = now - eventTime;
this.metrics.lastProcessedLsn = event.source.lsn;
this.metrics.lastEventTimestamp = new Date(eventTime);
// Track throughput over sliding window
this.eventTimestamps.push(now);
const windowStart = now - 60_000;
this.eventTimestamps = this.eventTimestamps.filter(
(t) => t > windowStart
);
this.metrics.eventsPerSecond =
this.eventTimestamps.length / 60;
}
checkHealth(): {
healthy: boolean;
alerts: string[];
} {
const alerts: string[] = [];
if (this.metrics.replicationLag > 30_000) {
alerts.push(
`High replication lag: ${this.metrics.replicationLag}ms`
);
}
const timeSinceLastEvent =
Date.now() - this.metrics.lastEventTimestamp.getTime();
if (timeSinceLastEvent > 300_000) {
alerts.push(
`No events for ${Math.round(timeSinceLastEvent / 1000)}s`
);
}
if (this.metrics.deadLetterCount > 100) {
alerts.push(
`${this.metrics.deadLetterCount} events in dead letter queue`
);
}
return {
healthy: alerts.length === 0,
alerts,
};
}
}Key Takeaways
Change Data Capture turns your database's transaction log into a real-time event stream without burdening the source with polling queries or trigger overhead. Debezium provides a production-grade connector that reads PostgreSQL WAL or MySQL binlog and publishes structured events to Kafka. Route events by table and operation type, handling creates, updates, and deletes with distinct logic for each downstream system. Plan for schema evolution from the start—register schema versions and define migration transforms that keep older events compatible with newer consumers. Enforce exactly-once processing through idempotency checks keyed on the event's log sequence number, preventing duplicate downstream effects on retries. Monitor replication lag continuously because a CDC pipeline that silently falls behind delivers stale data without warning, which is worse than having no pipeline at all.
