Real-Time Change Data Capture in Node.js: Streaming PostgreSQL Changes with Debezium and Kafka

Traditional event-driven architectures in Node.js often rely on application-layer event publishing. Your API receives a request, mutates the database, and then manually dispatches an event to a message broker like RabbitMQ, Redis, or Apache Kafka.

While this pattern works for simple CRUD applications, it introduces a dangerous flaw: the dual-write problem. If your database write succeeds, but the network drops before the event is published to your broker, your system drifts out of sync. Cache layers get stale, search indexes fall behind, and downstream microservices miss critical state changes.

To build truly reliable distributed systems, we need to eliminate application-layer event publishing entirely. We need Change Data Capture (CDC).

In this guide, we will build a production-ready real-time event streaming pipeline. We will use PostgreSQL as our source of truth, Debezium to read the Write-Ahead Log (WAL), Apache Kafka as our high-throughput event backbone, and a Node.js microservice to consume and process those changes in real time.

Architecture Overview

The pipeline we are building follows an end-to-end log-based CDC architecture:

+-------------------+      WAL / Replication      +--------------------+      Kafka Topic     +------------------+
|    PostgreSQL     | --------------------------> |  Debezium Connect  | -------------------> |   Apache Kafka   |
| (Source Database) |                             |  (Kafka Connector) |                      | (Event Streaming)|
+-------------------+                             +--------------------+                      +------------------+
                                                                                                                       |
                                                                                                                       v
                                                                                                              +------------------+
                                                                                                              |  Node.js Consumer|
                                                                                                              |   (kafkajs App)  |
                                                                                                              +------------------+

1. PostgreSQL Write-Ahead Log (WAL): Every insert, update, or delete operation is immutably appended to the PostgreSQL transaction log.
2. Debezium Connector: A distributed service running on Kafka Connect that continuously tails the PostgreSQL WAL using logical decoding plugins (pgoutput).
3. Apache Kafka: Receives structured JSON/Avro events from Debezium and partitions them by entity ID to guarantee ordered processing.
4. Node.js Consumer: A lightweight application utilizing kafkajs to read from the Kafka topic and react to data mutations.

Step 1: Configuring PostgreSQL for Logical Replication

By default, PostgreSQL does not emit granular row-level change events to its transaction log. We must configure it to use Logical Replication.

Update postgresql.conf

Ensure your PostgreSQL instance has the following settings enabled:

# Enable write-ahead log level for logical decoding
wal_level = logical

# Maximum number of concurrent connection slots for logical replication
max_replication_slots = 4

# Maximum number of WAL sender processes
max_wal_senders = 4

Create a Test Table and Replication User

Connect to your PostgreSQL instance and execute the following SQL commands to prepare our database environment:

-- Create a sample table
CREATE TABLE accounts (
    id SERIAL PRIMARY KEY,
    owner VARCHAR(100) NOT NULL,
    balance NUMERIC(12, 2) NOT NULL,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Create a dedicated user for Debezium
CREATE ROLE debezium_user WITH REPLICATION LOGIN PASSWORD 'secret_password';

-- Grant necessary permissions
GRANT SELECT ON TABLE accounts TO debezium_user;
GRANT USAGE ON SCHEMA public TO debezium_user;

Step 2: Deploying the Infrastructure (Docker Compose)

To orchestrate PostgreSQL, Apache Kafka, Zookeeper, and Debezium Kafka Connect locally, we will use Docker Compose. Create a docker-compose.yml file in your project root:

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_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1

  postgres:
    image: postgres:15-alpine
    ports:
      - "5432:5432"
    environment:
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: postgres
      POSTGRES_DB: cdc_demo
    command: postgres -c wal_level=logical -c max_replication_slots=4 -c max_wal_senders=4

  connect:
    image: debezium/connect:2.4
    ports:
      - "8083:8083"
    depends_on:
      - kafka
      - postgres
    environment:
      BOOTSTRAP_SERVERS: 'kafka:9092'
      GROUP_ID: 1
      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

Bring up the infrastructure by running:

docker-compose up -d

Step 3: Registering the Debezium PostgreSQL Connector

Once Kafka Connect is running (give it about 30 seconds to initialize), we need to register the PostgreSQL connector via its REST API. This tells Debezium which database to tail and which tables to capture.

Send a POST request to http://localhost:8083/connectors using curl or your favorite HTTP client:

curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" \
http://localhost:8083/connectors/ \
-d '{
  "name": "inventory-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "tasks.max": "1",
    "plugin.name": "pgoutput",
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "debezium_user",
    "database.password": "secret_password",
    "database.dbname": "cdc_demo",
    "database.server.name": "dbserver1",
    "table.include.list": "public.accounts",
    "publication.autocreate.mode": "all"
  }
}'

Verify that the connector is running successfully:

curl -s http://localhost:8083/connectors/inventory-connector/status

You should see a status of RUNNING for both the connector and its task.

Step 4: Building the Node.js Consumer

Now that Debezium is streaming database mutations straight into Kafka topics (specifically dbserver1.public.accounts), we can build a robust Node.js consumer using kafkajs.

Initialize Project and Install Dependencies

mkdir cdc-consumer
cd cdc-consumer
npm init -y
npm install kafkajs dotenv
npm install --save-dev typescript @types/node ts-node
npx tsc --init

Write the Consumer Implementation

Create a file named consumer.ts:

import { Kafka, EachMessagePayload } from 'kafkajs';

const kafka = new Kafka({
  clientId: 'account-service-consumer',
  brokers: [process.env.KAFKA_BROKER || 'localhost:9092'],
});

const consumer = kafka.consumer({ groupId: 'account-audit-group' });

interface DebeziumPayload {
  before: Record<string, any> | null;
  after: Record<string, any> | null;
  source: {
    version: string;
    connector: string;
    name: string;
    ts_ms: number;
    db: string;
    schema: string;
    table: string;
    txId: number;
    lsn: number;
  };
  op: 'c' | 'u' | 'd'; // Create, Update, Delete
  ts_ms: number;
}

async function run() {
  await consumer.connect();
  console.log('Kafka Consumer connected successfully.');

  // Subscribe to the Debezium topic for our accounts table
  await consumer.subscribe({ topic: 'dbserver1.public.accounts', fromBeginning: true });

  await consumer.run({
    eachMessage: async ({ topic, partition, message }: EachMessagePayload) => {
      if (!message.value) return;

      try {
        const parsedMessage = JSON.parse(message.value.toString());
        const payload: DebeziumPayload = parsedMessage.payload;

        if (!payload) return;

        const operationMap = {
          c: 'INSERT',
          u: 'UPDATE',
          d: 'DELETE',
        };

        const operation = operationMap[payload.op] || 'UNKNOWN';

        console.log(`
-----------------------------------------`);
        console.log(`[CDC Event Captured] Table: ${payload.source.table}`);
        console.log(`Operation: ${operation}`);
        console.log(`Timestamp: ${new Date(payload.ts_ms).toISOString()}`);
        
        if (operation === 'INSERT') {
          console.log('New Row Data:', payload.after);
        } else if (operation === 'UPDATE') {
          console.log('Previous Data:', payload.before);
          console.log('Updated Data:', payload.after);
        } else if (operation === 'DELETE') {
          console.log('Deleted Row Data:', payload.before);
        }
        console.log(`-----------------------------------------`);

      } catch (error) {
        console.error('Failed to parse incoming Kafka message:', error);
      }
    },
  });
}

run().catch((err) => {
  console.error('Error running Kafka consumer:', err);
  process.exit(1);
});

Run your TypeScript consumer:

npx ts-node consumer.ts

Step 5: Testing the Pipeline

Let’s test our end-to-end pipeline. Open a terminal and connect to your PostgreSQL database:

docker exec -it cdc_demo-postgres-1 psql -U postgres -d cdc_demo

1. Insert an Record

Execute an INSERT statement:

INSERT INTO accounts (owner, balance) VALUES ('Alice Smith', 1500.00);

Node.js Console Output:

-----------------------------------------
[CDC Event Captured] Table: accounts
Operation: INSERT
Timestamp: 202X-10-24T12:00:00.000Z
New Row Data: { id: 1, owner: 'Alice Smith', balance: 1500, updated_at: '...' }
-----------------------------------------

2. Update a Record

Execute an UPDATE statement:

UPDATE accounts SET balance = 1750.50 WHERE owner = 'Alice Smith';

Node.js Console Output:

-----------------------------------------
[CDC Event Captured] Table: accounts
Operation: UPDATE
Timestamp: 202X-10-24T12:05:00.000Z
Previous Data: { id: 1, owner: 'Alice Smith', balance: 1500, updated_at: '...' }
Updated Data: { id: 1, owner: 'Alice Smith', balance: 1750.5, updated_at: '...' }
-----------------------------------------

3. Delete a Record

Execute a DELETE statement:

DELETE FROM accounts WHERE owner = 'Alice Smith';

Node.js Console Output:

-----------------------------------------
[CDC Event Captured] Table: accounts
Operation: DELETE
Timestamp: 202X-10-24T12:10:00.000Z
Deleted Row Data: { id: 1, owner: 'Alice Smith', balance: 1750.5, updated_at: '...' }
-----------------------------------------

Production Considerations

Moving a CDC pipeline from a local development environment to production requires careful planning around operational edge cases:

• WAL Disk Space and Retention: If your downstream consumers go offline for an extended period, PostgreSQL will continue retaining WAL segments to prevent data loss. Monitor your disk usage closely and configure appropriate replication slot timeouts.
• Schema Evolution: When you alter database tables (e.g., ALTER TABLE accounts ADD COLUMN email VARCHAR(255)), Debezium emits schema changes alongside row data. Ensure your Node.js consumers are written defensively to handle dynamic or evolving payloads gracefully.
• Idempotency: Kafka guarantees at-least-once delivery semantics by default. Your Node.js event handlers must be idempotent—capable of processing the exact same mutation event multiple times without side effects (e.g., using UPSERT logic or event deduplication tables).

Conclusion

By migrating away from application-layer event publishing to a log-based Change Data Capture pipeline with PostgreSQL, Debezium, and Kafka, you decouple your business logic from data distribution. Your Node.js microservices gain absolute confidence that every single database mutation is reliably captured, ordered, and streamed in real time.