Debezium PostgreSQL Connector — CDC in Action

Debezium is a distributed CDC platform that tails PostgreSQL’s Write-Ahead Log and produces a Kafka event for every row change. It runs as a connector inside Kafka Connect, which means you deploy it by POSTing a JSON config to a REST API — no custom code needed. This post walks through setting up a local Debezium stack, registering a connector, and reading the events it produces.

How It Works

PostgreSQL WAL (logical replication)
        |
        v
  Debezium Connector (runs inside Kafka Connect)
        |
        v
  Kafka Topics (one per table by default: <prefix>.<schema>.<table>)

Debezium uses PostgreSQL’s logical decoding. When you register a connector:

1. It creates a replication slot in PostgreSQL — a cursor that tracks how far Debezium has read in the WAL
2. It creates a publication (or uses an existing one) — tells PostgreSQL which tables to include
3. On first start, it does an initial snapshot — reads all existing rows and produces them as events with op: "r" (read)
4. After the snapshot, it switches to streaming — tailing the WAL in real time for INSERTs, UPDATEs, and DELETEs

The Stack

A minimal local stack needs five services: PostgreSQL (with wal_level=logical), Zookeeper, Kafka, Kafka Connect with the Debezium plugin, and optionally Confluent Control Center for a web UI.

A docker-compose.yml for this:

services:
  postgres:
    image: postgres:17.6-alpine
    environment:
      POSTGRES_DB: learn
      POSTGRES_PASSWORD: postgres
    command: [
      "postgres",
      "-c", "wal_level=logical",
      "-c", "max_replication_slots=4",
      "-c", "max_wal_senders=4"
    ]
    ports:
      - "5432:5432"
    volumes:
      - ./init.sql:/docker-entrypoint-initdb.d/init.sql

  zookeeper:
    image: confluentinc/cp-zookeeper:7.5.1
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181

  broker:
    image: confluentinc/cp-kafka:7.5.1
    depends_on: [zookeeper]
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,HOST:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:29092,HOST://localhost:9092
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

  connect:
    image: confluentinc/cp-server-connect:7.5.1
    depends_on: [broker]
    ports:
      - "8083:8083"
    environment:
      CONNECT_BOOTSTRAP_SERVERS: broker:29092
      CONNECT_REST_PORT: 8083
      CONNECT_GROUP_ID: connect-cluster
      CONNECT_CONFIG_STORAGE_TOPIC: _connect-configs
      CONNECT_OFFSET_STORAGE_TOPIC: _connect-offsets
      CONNECT_STATUS_STORAGE_TOPIC: _connect-status
      CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_KEY_CONVERTER: org.apache.kafka.connect.storage.StringConverter
      CONNECT_VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter
      CONNECT_VALUE_CONVERTER_SCHEMAS_ENABLE: "false"
      CONNECT_REST_ADVERTISED_HOST_NAME: connect
    command:
      - bash
      - -c
      - |
        confluent-hub install --no-prompt debezium/debezium-connector-postgresql:2.5.4
        /etc/confluent/docker/run

The key PostgreSQL setting is wal_level=logical — without it, the WAL does not contain enough information for logical decoding. The init.sql script creates a sample schema and seeds some rows so there is data to snapshot.

A sample init.sql:

CREATE TABLE orders (
    id SERIAL PRIMARY KEY,
    customer VARCHAR(255),
    product VARCHAR(255),
    amount NUMERIC(10,2),
    status VARCHAR(50) DEFAULT 'pending',
    created_at TIMESTAMP DEFAULT now(),
    updated_at TIMESTAMP DEFAULT now()
);

CREATE TABLE customers (
    id SERIAL PRIMARY KEY,
    name VARCHAR(255),
    email VARCHAR(255),
    tier VARCHAR(50) DEFAULT 'standard',
    created_at TIMESTAMP DEFAULT now()
);

INSERT INTO orders (customer, product, amount) VALUES
  ('alice', 'Widget A', 29.99),
  ('bob', 'Widget B', 49.99),
  ('carol', 'Widget C', 19.99);

INSERT INTO customers (name, email, tier) VALUES
  ('Alice', 'alice@example.com', 'premium'),
  ('Bob', 'bob@example.com', 'standard');

Registering a Connector

Once the stack is running and Kafka Connect is healthy (poll GET http://localhost:8083/ until it responds), register a connector by POSTing its configuration:

curl -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "name": "learn-cdc-connector",
    "config": {
      "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
      "database.hostname": "postgres",
      "database.port": "5432",
      "database.user": "postgres",
      "database.password": "postgres",
      "database.dbname": "learn",
      "topic.prefix": "learn",
      "plugin.name": "pgoutput",
      "slot.name": "learn_cdc_slot",
      "publication.autocreate.mode": "all_tables",
      "snapshot.mode": "initial",
      "tombstones.on.delete": "false",
      "key.converter": "org.apache.kafka.connect.storage.StringConverter",
      "value.converter": "org.apache.kafka.connect.json.JsonConverter",
      "value.converter.schemas.enable": "false"
    }
  }'

Config	What it does
plugin.name: pgoutput	Uses PostgreSQL’s built-in logical decoding output plugin
topic.prefix: learn	Topics are named <prefix>.<schema>.<table>, e.g. learn.public.orders
slot.name	The replication slot name in PostgreSQL — persists even if the connector is deleted
publication.autocreate.mode: all_tables	Debezium auto-creates a publication covering all tables
snapshot.mode: initial	On first start, snapshot all existing rows, then switch to streaming

Check the connector is running:

curl http://localhost:8083/connectors/learn-cdc-connector/status | jq

Both the connector and its task should show "state": "RUNNING".

Check what Debezium created in PostgreSQL:

SELECT slot_name, plugin, slot_type, active, restart_lsn
FROM pg_replication_slots;

SELECT * FROM pg_publication;
SELECT * FROM pg_publication_tables;

The replication slot learn_cdc_slot and an auto-created publication dbz_publication covering all tables — these are the two PostgreSQL objects Debezium needs for CDC.

The Initial Snapshot

List topics to see what Debezium created:

kafka-topics --bootstrap-server localhost:9092 --list

Topics like learn.public.orders and learn.public.customers appear. Consume from one:

kafka-console-consumer \
  --bootstrap-server localhost:9092 \
  --topic learn.public.orders \
  --from-beginning \
  --property print.key=true \
  --property key.separator=" | "

Events appear for the 3 seeded rows. Each event has this structure:

{
  "before": null,
  "after": {
    "id": 1,
    "customer": "alice",
    "product": "Widget A",
    "amount": 29.99,
    "status": "pending",
    "created_at": 1715500000000,
    "updated_at": 1715500000000
  },
  "source": {
    "version": "2.5.4.Final",
    "connector": "postgresql",
    "name": "learn",
    "ts_ms": 1715500000000,
    "snapshot": "first",
    "db": "learn",
    "schema": "public",
    "table": "orders",
    "txId": 1234,
    "lsn": 12345678
  },
  "op": "r",
  "ts_ms": 1715500000000
}

The Event Envelope

Every Debezium event has this structure:

Field	What it is
before	Row state before the change. null for snapshots and INSERTs. Only populated for UPDATEs/DELETEs when replica identity is FULL
after	Row state after the change. null for DELETEs
source	Metadata: database, schema, table, transaction ID, WAL position (LSN), whether this was a snapshot
op	The operation type
ts_ms	Timestamp when Debezium processed the event

Operation types:

op	Meaning	When
r	read	Initial snapshot — existing rows
c	create	INSERT after streaming starts
u	update	UPDATE
d	delete	DELETE (after is null, before has the deleted row if replica identity is FULL)

The seeded rows appear as op: "r" because the INSERTs in init.sql happened before Debezium connected. When the connector starts with snapshot.mode: initial, it reads all existing rows and publishes them as read events. Changes made after the connector is running are captured from the WAL as c/u/d.

Live Changes

With the consumer still running, open a psql session and make changes:

-- INSERT: produces an op: "c" event
INSERT INTO orders (customer, product, amount)
VALUES ('dave', 'Widget D', 39.99);

-- UPDATE: produces an op: "u" event
UPDATE orders SET status = 'shipped' WHERE id = 1;

-- DELETE: produces an op: "d" event
DELETE FROM orders WHERE id = 3;

Three new events appear on the consumer:

• The INSERT has "op": "c" with before: null and after containing the new row
• The UPDATE has "op": "u" with before: null and after with the updated row — before is null because the default replica identity only includes the primary key in WAL records, not the full row
• The DELETE has "op": "d" with after: null — and before is also null for the same reason

Message Keys

The key printed before the | separator is the table’s primary key, serialized as JSON:

{"id":1} | {"before": null, "after": {...}, "op": "u", ...}

Debezium uses the primary key as the Kafka message key, which means all events for the same row go to the same Kafka partition — guaranteeing ordering per row.

What Debezium Reads vs. What the Application Reads

Debezium reads the WAL, not the tables themselves. It opens a replication connection to PostgreSQL (using replication=database in the connection string) and receives a stream of decoded changes. This is the same mechanism PostgreSQL uses for built-in logical replication between database instances. The application’s regular queries are unaffected — Debezium adds no load to the query path. The only overhead is the WAL retention: PostgreSQL must hold WAL segments that the replication slot has not yet consumed.