PostgreSQL Logical Replication to Kafka: End-to-End Setup

Running PostgreSQL at scale means your application writes are only half the story — the real challenge is getting those changes downstream to analytics pipelines, search indexes, and microservices without burning your primary under the load of polling queries. Logical replication, introduced in PostgreSQL 10 and matured significantly since, gives you a structured change stream directly from the write-ahead log. Paired with Debezium and Kafka, you get a durable, replayable event stream that decouples your producers from every downstream consumer. This guide walks you through a production-grade end-to-end setup, covering the PostgreSQL side, Debezium connector configuration, topic structure, lag monitoring, and schema evolution with Confluent Schema Registry.

Architecture Overview

The logical replication pipeline has four distinct layers that each need to be configured correctly for the whole thing to work reliably.

PostgreSQL writes every row change to the write-ahead log (WAL) first, before applying it to the heap. With wal_level = logical, the WAL entries include enough information — column-level old and new values — for an external decoder to reconstruct the logical meaning of each change. A replication slot tracks how far a downstream consumer has processed the WAL, ensuring PostgreSQL retains WAL segments until they are acknowledged. A publication is a named filter that declares which tables (and which operations) are included in the stream.

Debezium is a Kafka Connect source connector that acts as a PostgreSQL replication client. It connects via the replication protocol, receives decoded change events from the slot, and publishes them as Kafka messages. Kafka then acts as the durable log — consumers can replay from any offset, and you can fan out to as many downstream systems as you need without touching PostgreSQL again.

Schema Registry sits alongside Kafka and stores Avro schemas for each topic. Producers register a schema on first write; consumers fetch and cache it by ID. This gives you schema evolution with backward/forward compatibility guarantees, and prevents a bad producer from breaking all consumers with an incompatible change.

PostgreSQL Setup

Start by verifying and setting the WAL level. This requires a PostgreSQL restart, so plan accordingly on production systems.

-- Check current setting
SHOW wal_level;

-- If not already 'logical', set it in postgresql.conf:
-- wal_level = logical
-- Then restart PostgreSQL.

You also need to raise max_replication_slots and max_wal_senders if you plan to run multiple connectors. A safe starting point for most setups:

# postgresql.conf
wal_level = logical
max_replication_slots = 10
max_wal_senders = 10

Next, create a dedicated replication user. Never use a superuser for this in production — scope it precisely:

CREATE ROLE debezium_replication WITH
  REPLICATION
  LOGIN
  PASSWORD 'use-a-strong-secret-here';

-- Grant SELECT on tables you want to replicate
GRANT SELECT ON ALL TABLES IN SCHEMA public TO debezium_replication;
ALTER DEFAULT PRIVILEGES IN SCHEMA public
  GRANT SELECT ON TABLES TO debezium_replication;

Now create the publication. A publication is your authoritative declaration of what gets streamed. You can include all tables or enumerate them explicitly — explicit is safer in production because it prevents accidentally streaming a table you add later:

-- Stream specific tables
CREATE PUBLICATION debezium_pub
  FOR TABLE public.orders, public.customers, public.products
  WITH (publish = 'insert, update, delete');

-- Or stream everything in a schema (use with caution)
-- CREATE PUBLICATION debezium_pub FOR ALL TABLES;

Finally, create the replication slot using the pgoutput plugin. The pgoutput plugin is built into PostgreSQL 10+ and requires no additional installation, unlike the older wal2json plugin:

SELECT pg_create_logical_replication_slot(
  'debezium_slot',   -- slot name, referenced in connector config
  'pgoutput'         -- built-in logical decoder plugin
);

Debezium PostgreSQL Connector

Deploy Kafka Connect with the Debezium PostgreSQL connector plugin on the classpath. The connector configuration is submitted as a JSON payload to the Kafka Connect REST API:

curl -X POST http://kafka-connect:8083/connectors \
  -H "Content-Type: application/json" \
  -d @debezium-postgres.json

{
  "name": "postgres-debezium-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "tasks.max": "1",

    "database.hostname": "postgres.internal",
    "database.port": "5432",
    "database.user": "debezium_replication",
    "database.password": "${file:/opt/secrets/db.properties:password}",
    "database.dbname": "production",
    "database.server.name": "prod_pg",

    "plugin.name": "pgoutput",
    "slot.name": "debezium_slot",
    "publication.name": "debezium_pub",

    "table.include.list": "public.orders,public.customers,public.products",

    "topic.prefix": "prod_pg",

    "heartbeat.interval.ms": "10000",
    "heartbeat.action.query": "UPDATE debezium_heartbeat SET ts = now() WHERE id = 1",

    "decimal.handling.mode": "double",
    "time.precision.mode": "connect",
    "interval.handling.mode": "string",

    "snapshot.mode": "initial",

    "key.converter": "io.confluent.connect.avro.AvroConverter",
    "key.converter.schema.registry.url": "http://schema-registry:8081",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://schema-registry:8081"
  }
}

Create the heartbeat table and add it to the publication:

CREATE TABLE debezium_heartbeat (
  id   INT PRIMARY KEY,
  ts   TIMESTAMPTZ NOT NULL DEFAULT now()
);
INSERT INTO debezium_heartbeat VALUES (1, now());

-- Add it to the existing publication
ALTER PUBLICATION debezium_pub ADD TABLE debezium_heartbeat;

Topic Structure and Message Envelope

Debezium creates Kafka topics automatically using the pattern {topic.prefix}.{schema}.{table}. With the configuration above, your topics will be:

prod_pg.public.orders
prod_pg.public.customers
prod_pg.public.products
prod_pg.public.debezium_heartbeat

Each message follows the Debezium envelope format. The key contains the table's primary key column(s); the value contains the full change event:

{
  "schema": { ... },
  "payload": {
    "before": {
      "id": 1042,
      "status": "pending",
      "total": 149.99
    },
    "after": {
      "id": 1042,
      "status": "shipped",
      "total": 149.99
    },
    "source": {
      "version": "2.5.0.Final",
      "connector": "postgresql",
      "name": "prod_pg",
      "ts_ms": 1708540800000,
      "snapshot": "false",
      "db": "production",
      "schema": "public",
      "table": "orders",
      "txId": 7392841,
      "lsn": 24589123456,
      "xmin": null
    },
    "op": "u",
    "ts_ms": 1708540800123
  }
}

The op field identifies the operation type: c for insert (create), u for update, d for delete, and r for read (snapshot). For delete events, after is null; for insert events, before is null. Your consumers must handle all four cases. The source.lsn field gives you the WAL position, which you can use for deduplication or ordering guarantees.

Monitoring Replication Slot Lag

Replication slot lag is the most operationally dangerous aspect of this setup. If Debezium falls behind — or goes offline entirely — PostgreSQL cannot reclaim WAL segments that the slot has not yet consumed. Your disk will fill up and PostgreSQL will stop accepting writes.

Query slot lag in bytes and LSN distance:

SELECT
  slot_name,
  plugin,
  active,
  pg_size_pretty(
    pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn)
  ) AS lag_bytes_pretty,
  pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn) AS lag_bytes,
  confirmed_flush_lsn,
  pg_current_wal_lsn() AS current_lsn
FROM pg_replication_slots
WHERE slot_type = 'logical'
ORDER BY lag_bytes DESC;

Set up alerting at two thresholds: warn when lag exceeds 5 GB and page when it exceeds 20 GB (adjust for your disk capacity and WAL generation rate). Also alert when active = false — an inactive slot is accumulating WAL with no consumer making progress.

Also monitor Kafka consumer lag from the Debezium connector's output side using the Kafka consumer group offset:

kafka-consumer-groups.sh \
  --bootstrap-server kafka:9092 \
  --describe \
  --group your-downstream-consumer-group

PostgreSQL slot lag and Kafka consumer group lag are two independent metrics — both need monitoring. A fast Debezium connector that is keeping up with PostgreSQL does not help if your downstream consumer is 10 million messages behind in Kafka.

Schema Registry and Avro

Using Avro with Confluent Schema Registry gives you schema evolution with enforced compatibility rules. When Debezium writes to a topic for the first time, the Avro converter registers the schema derived from the table's column definitions. Each subsequent message is serialized with a 5-byte header containing a magic byte and the schema ID, followed by the Avro binary payload.

Schema Registry compatibility modes determine what changes are allowed:

# Set BACKWARD compatibility for a topic's value schema
# (new schema can read data written with old schema)
curl -X PUT \
  http://schema-registry:8081/config/prod_pg.public.orders-value \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d '{"compatibility": "BACKWARD"}'

The compatibility matrix for common column changes:

• Adding a nullable column — backward compatible; add with a default of null in the Avro schema.
• Adding a NOT NULL column without default — breaking change; requires a full snapshot or explicit schema migration strategy.
• Dropping a column — forward compatible only; consumers reading old messages will see the field as its default value.
• Changing a column type — almost always breaking; requires a new topic or a schema version bump with a union type.

To inspect registered schemas and check for compatibility before a migration:

# List all schema versions for a subject
curl http://schema-registry:8081/subjects/prod_pg.public.orders-value/versions

# Check if a new schema is compatible
curl -X POST \
  http://schema-registry:8081/compatibility/subjects/prod_pg.public.orders-value/versions/latest \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d @new-schema.json

Stream Your PostgreSQL Changes with JusDB

Setting up logical replication, managing slot lag, and operating Schema Registry is a significant operational surface area — and that is before you factor in connector upgrades, Kafka cluster management, and schema migration runbooks. JusDB provides managed PostgreSQL with built-in change data capture support, pre-configured replication monitoring, and alerting that catches slot lag before it becomes a disk-full incident.

If you are evaluating this stack for a new project or looking to reduce the operational burden on an existing pipeline, explore JusDB's managed database platform and see how much of this infrastructure you can hand off.