CDC Connector Setup for Postgres

Postgres change data capture is logical replication with a consumer that happens to write to a search index instead of a replica. Everything that makes it reliable — and everything that makes it dangerous — lives in the replication slot: it guarantees no change is missed, and it will fill your disk if the consumer stops. This page configures the connector end to end and then makes the slot safe to operate. It is part of change data capture setup inside the data ingestion and synchronization pipelines area, and it is the Postgres counterpart to the binlog-based MySQL connector setup.

Prerequisites

1. PostgreSQL 12 or later with wal_level = logical — this requires a restart, so plan it.
2. A role with the REPLICATION attribute and SELECT on the tables being captured.
3. Kafka Connect with the Debezium Postgres connector, or an equivalent consumer that can hold a slot.
4. Disk headroom on the database host equal to at least a few hours of WAL at peak write rate.
5. A primary key (or REPLICA IDENTITY FULL) on every captured table, or updates will arrive without their old values.

Diagnosis: the slot is the contract, and it is one-sided

A replication slot tells Postgres “do not discard WAL past this position, because a consumer still needs it.” That is exactly the guarantee you want — a consumer restart loses nothing — and it comes with an obligation the database cannot enforce: if the consumer never advances, WAL accumulates forever.

The incident this causes is always the same, and it is a database outage rather than a search outage:

$ SELECT slot_name, active, pg_size_pretty(
      pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS retained
  FROM pg_replication_slots;
 slot_name        | active | retained
------------------+--------+----------
 search_cdc_slot  | f      | 214 GB      <- consumer down since Friday; disk is 89% full

The slot is inactive, so nothing is consuming, and Postgres is faithfully keeping 214 GB of WAL for a consumer that may never come back. Any Postgres CDC setup that does not monitor slot lag from day one is an outage waiting for a long weekend.

Solution Steps

1. Enable logical decoding and size the WAL guardrails

wal_level = logical requires a restart. Set the slot guardrail at the same time so a stalled consumer degrades into a broken slot rather than a full disk.

# postgresql.conf
wal_level = logical
max_replication_slots = 8          # one per consumer, plus headroom
max_wal_senders = 8
# Postgres 13+: cap what a single slot may retain. Beyond this the slot is
# invalidated (recoverable by re-snapshotting) instead of filling the volume.
max_slot_wal_keep_size = 64GB

2. Create a least-privilege role and a scoped publication

Capture only the tables the index needs. A FOR ALL TABLES publication couples your search pipeline to every future schema change in the database.

CREATE ROLE cdc_search WITH REPLICATION LOGIN PASSWORD 'change-me';
GRANT USAGE ON SCHEMA public TO cdc_search;
GRANT SELECT ON public.products, public.product_variants TO cdc_search;

-- Explicit table list: adding a table to the index becomes a deliberate change.
CREATE PUBLICATION search_pub FOR TABLE public.products, public.product_variants;

-- Updates must carry enough identity for the indexer to address the document.
ALTER TABLE public.products REPLICA IDENTITY USING INDEX products_pkey;

3. Configure the connector against pgoutput

pgoutput is built into Postgres and needs no extension, which matters on managed instances where installing wal2json is not an option.

{
  "name": "products-postgres-cdc",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "localhost",
    "database.port": "5432",
    "database.user": "cdc_search",
    "database.dbname": "shop",
    "plugin.name": "pgoutput",
    "publication.name": "search_pub",
    "publication.autocreate.mode": "disabled",
    "slot.name": "search_cdc_slot",
    "slot.drop.on.stop": "false",
    "table.include.list": "public.products,public.product_variants",
    "snapshot.mode": "initial",
    "heartbeat.interval.ms": "10000",
    "topic.prefix": "shop"
  }
}

Two settings carry outsized weight. heartbeat.interval.ms makes the connector advance the slot even when the captured tables are idle — without it, a busy database with a quiet products table retains WAL forever because the consumer never acknowledges a position. slot.drop.on.stop: false keeps the slot across restarts, which is what you want in production and what makes cleanup a deliberate operation.

4. Take the initial snapshot without blocking writers

snapshot.mode: initial reads a consistent view of each table before streaming. On large tables use the incremental snapshot so the export runs in chunks alongside streaming instead of as one long transaction.

{
  "snapshot.mode": "initial",
  "incremental.snapshot.chunk.size": "8192",
  "signal.data.collection": "public.debezium_signal"
}

-- The signalling table lets you trigger or resume a snapshot without a restart.
CREATE TABLE public.debezium_signal (
  id    TEXT PRIMARY KEY,
  type  TEXT NOT NULL,
  data  TEXT NULL
);
GRANT INSERT, SELECT ON public.debezium_signal TO cdc_search;

5. Map change events onto index operations

Postgres deletes arrive as an event with after: null followed by a tombstone. Translate the three operations explicitly; treating every event as an upsert is how deleted products keep appearing in search.

# apply_events.py — Debezium envelope to index operations
def to_index_op(evt: dict):
    payload = evt["payload"]
    op = payload["op"]                       # c=create, u=update, d=delete, r=snapshot read
    if op == "d":
        return {"delete": {"_index": "products", "_id": payload["before"]["id"]}}
    row = payload["after"]
    return {"index": {"_index": "products", "_id": row["id"]}}, {
        "title": row["title"],
        "price": float(row["price"]),
        # Carry the LSN so out-of-order retries can be rejected by external versioning.
        "_version_lsn": payload["source"]["lsn"],
    }

6. Monitor slot lag as a first-class alert

This is the step that separates a working demo from a production connector.

-- Alert when a slot retains more than 16 GB or has been inactive for 15 minutes.
SELECT slot_name,
       active,
       pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn) AS retained_bytes
FROM pg_replication_slots
WHERE NOT active
   OR pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn) > 16 * 1024^3;

Verification

Confirm the slot exists, is active, and is not accumulating:

psql -U cdc_search -d shop -c "SELECT slot_name, plugin, active, \
  pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS retained \
  FROM pg_replication_slots WHERE slot_name = 'search_cdc_slot';"
#  slot_name       | plugin   | active | retained
# -----------------+----------+--------+----------
#  search_cdc_slot | pgoutput | t      | 1264 kB

Confirm a real row change reaches the index end to end:

psql -U postgres -d shop -c "UPDATE products SET title = 'Trail Runner 4' WHERE id = 991;"
sleep 3
curl -s 'localhost:9200/products/_doc/991?filter_path=_source.title'
# => {"_source":{"title":"Trail Runner 4"}}

Confirm deletes remove the document rather than leaving a stale copy:

psql -U postgres -d shop -c "DELETE FROM products WHERE id = 991;"
sleep 3
curl -s -o /dev/null -w '%{http_code}\n' 'localhost:9200/products/_doc/991'
# => 404

Common Pitfalls

Operational notes

Two Postgres-specific behaviours will eventually surprise you. The first is that logical decoding happens on the primary only: a physical replica cannot host a logical slot in most deployments, so CDC load lands on the same instance serving your application writes. Decoding is cheap relative to query execution, but the WAL retention is not free, and on a small instance the extra disk and the decoding backend are worth budgeting explicitly rather than discovering.

The second is schema evolution. Adding a nullable column is invisible to the connector until the first change event carries it, and the event’s schema block changes shape at that moment. Consumers that deserialize into a fixed structure break precisely then — usually hours after the migration, when the first row on that table is updated, which makes the correlation hard to see. Handle it by consuming the envelope defensively (read fields by name, tolerate unknown ones) and by treating captured tables as a published interface: a migration that touches one is a coordinated change, not a routine one. The same discipline is covered generally in handling schema evolution in CDC streams.

One more sizing note. Logical decoding buffers each transaction until it commits, so a single very large transaction — a bulk UPDATE touching millions of rows — materialises as one enormous burst of change events after the commit, not as a stream during it. Bulk maintenance on a captured table should therefore be chunked into transactions of a few thousand rows, both to keep the decoding memory bounded and to avoid handing your indexer a multi-gigabyte spike that trips the backpressure controls described in handling 429 rejections with backpressure.

Related

• Change data capture setup — the parent guide covering CDC architecture and delivery semantics across databases.
• CDC connector setup for MySQL — the binlog-based equivalent, with GTID positioning instead of slots.
• Bulk indexing throughput tuning — how the consumer should write the change events it decodes.