PostgreSQL Debezium CDC Backlog Sentinel

Audit PostgreSQL logical replication streams using Debezium, persist state offsets in Kestra Key-Value store, evaluate event accumulation against operational thresholds, and dispatch alerts to Slack when change backlogs threaten database stability.

How it works

1. Probe and Stream: The capture_cdc_events task (io.kestra.plugin.debezium.postgres.Capture) establishes a logical replication stream with PostgreSQL using the native PGOUTPUT plugin. It consumes pending change data capture records up to inputs.max_records within a 10-second polling window (maxWait: PT10S). The task outputs the captured row count (outputs.capture_cdc_events.size), stores the change events in internal storage (outputs.capture_cdc_events.uris), and atomically records the Log Sequence Number (LSN) position into Kestra KV store (orders-cdc-checkpoint).
2. Evaluate Threshold: The evaluate_cdc_backlog task (io.kestra.plugin.core.flow.Switch) inspects the returned change count against inputs.backlog_threshold.
3. Conditional Action:
   • If the count meets or exceeds the threshold ("true"), check_dry_run evaluates inputs.dry_run. When dry_run: "false", alert_backlog_threshold_breach dispatches a Slack alert via io.kestra.plugin.slack.notifications.SlackIncomingWebhook reporting the exact row count, database name, and offset checkpoint key; when dry_run: "true", log_dry_run_breach records a dry-run observation without sending alerts.
   • If the stream is healthy ("false"), log_compliant_cdc_stream records an operational log confirming normal CDC velocity.
4. Governance Summary: log_governance_cycle_complete logs the final execution summary with internal storage references.
5. Error Containment: A flow-level errors block dispatches an alert to Slack if the database is unreachable or replication credentials expire.

What you get

• Non-intrusive micro-batch inspection of PostgreSQL logical decoding buffers without external Kafka Connect clusters.
• Automated detection of unconsumed WAL change accumulation before PostgreSQL replication slots consume disk storage.
• Persistent LSN offset checkpointing backed by Kestra KV store for consistent restart capability.
• Automated Slack alerting on anomalous change spikes or consumer processing delays.

Who it is for

• Data Platform Engineers operating PostgreSQL CDC pipelines feeding downstream lakes and warehouses.
• Database Administrators monitoring WAL growth and replication slot health.
• Site Reliability Engineers safeguarding database disk headroom against runaway logical decoding backlogs.

Why orchestrate this with Kestra

Debezium connectors typically require heavy Kafka Connect infrastructure, schema registries, and consumer daemons just to inspect changes. Kestra packages the Debezium engine directly as a first-class task. Orchestrating CDC through Kestra provides declarative scheduling, built-in retry policies, native KV-backed offset tracking, conditional branching, and enterprise Slack alerting within a single YAML definition.

Prerequisites

• A running PostgreSQL instance (version 12+) configured with wal_level = logical.
• A database user with REPLICATION role granted or table owner permissions.
• A publication covering the targeted tables (defaults to auto-managed PGOUTPUT publication).
• Network connectivity from the Kestra worker to the PostgreSQL port (default 5432).
• Slack incoming webhook URL for notifications.

Secrets

Configure the following secrets in Kestra before running the flow:

• POSTGRES_PASSWORD: Password for the PostgreSQL replication user account.
• SLACK_WEBHOOK: Complete Slack incoming webhook URL for notifications.

For open-source Kestra instances, export secrets as environment variables:

export SECRET_POSTGRES_PASSWORD="<your_postgres_password>"
export SECRET_SLACK_WEBHOOK="https://hooks.slack.com/services/<WORKSPACE>/<CHANNEL>/<TOKEN>"

Inputs

Name	Type	Default	Allowed Values	Description
hostname	STRING	postgres-cdc	Any valid hostname	PostgreSQL host or container name.
port	STRING	"5432"	Port string	Port number for PostgreSQL server.
database	STRING	orders_db	Database identifier	PostgreSQL database name.
username	STRING	postgres	User identifier	Replication database user.
included_tables	STRING	public.orders	Schema.table pattern	Target tables to stream CDC from.
max_records	INT	500	Positive integer	Maximum records to probe per cycle.
backlog_threshold	INT	100	Positive integer	Count threshold triggering Slack alert.
dry_run	SELECT	"true"	"true", "false"	Observation mode flag.

Outputs

The workflow exposes the following task outputs for downstream inspection or parent flow orchestration:

• {{ outputs.capture_cdc_events.size }}: Total number of CDC events captured in the current cycle.
• {{ outputs.capture_cdc_events.uris }}: Kestra internal storage URIs map pointing to the serialized Ion record streams.
• {{ outputs.capture_cdc_events.stateOffsetKey }}: Kestra KV store key holding the saved LSN offset map.

Quick start

1. Launch a Local PostgreSQL Container with Logical Decoding

Start an isolated PostgreSQL container configured with logical decoding enabled on non-default port 5433 to avoid host port 8080 conflicts:

docker run -d \
  --name postgres-cdc \
  -e POSTGRES_PASSWORD=postgres \
  -e POSTGRES_DB=orders_db \
  -p 5433:5432 \
  postgres:16-alpine \
  -c wal_level=logical \
  -c max_replication_slots=4 \
  -c max_wal_senders=4

Connect the container to your Kestra Docker network (find the network name with docker network ls):

docker network connect YOUR_KESTRA_NETWORK postgres-cdc

2. Initialize Seed Table and Change Records

Create a table and seed records to generate change events:

docker exec -i postgres-cdc psql -U postgres -d orders_db << 'EOF'
CREATE TABLE IF NOT EXISTS public.orders (
    id SERIAL PRIMARY KEY,
    customer_id VARCHAR(50),
    amount NUMERIC(10, 2),
    status VARCHAR(20),
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

INSERT INTO public.orders (customer_id, amount, status)
VALUES ('CUST-100', 49.99, 'PENDING'), ('CUST-101', 129.50, 'PAID');
EOF

3. Run the Workflow in Kestra

Trigger the flow manually with dry_run: "true". Inspect the execution overview to verify that outputs.capture_cdc_events.size reflects the newly created rows and that orders-cdc-checkpoint is committed in Kestra KV store.

How to extend

• Micro-batch Lakehouse Ingestion: Append downstream tasks such as io.kestra.plugin.duckdb.Query or io.kestra.plugin.jdbc.postgresql.Query inside the "true" branch to transform and append captured change records into an analytics data lake.
• Replication Slot Disk Guard: Combine this flow with a periodic query against pg_replication_slots to monitor active status and wal_status alongside CDC volume.
• Schema Change Notification: Inspect Debezium metadata fields for DDL events and trigger downstream dbt model runs when schema alterations are detected.

Documentation

• Kestra Debezium PostgreSQL Plugin
• Kestra Slack Notification Plugin
• Debezium PostgreSQL Connector Documentation

Pitfalls

• WAL Level Configuration: PostgreSQL defaults to wal_level = replica. Debezium requires wal_level = logical. Setting this requires restarting PostgreSQL.
• Replication Slot Growth: Logical replication slots prevent PostgreSQL from purging WAL files until the consumer acknowledges them. Ensure the flow runs on a regular schedule or cleans up stale slots so PostgreSQL disk space is preserved.
• Superuser Privileges: The connector user requires the REPLICATION attribute in PostgreSQL (ALTER USER <username> WITH REPLICATION;) or membership in pg_read_all_data.
• Docker Host Networking: When running Kestra and PostgreSQL in separate Docker networks, use container DNS names or IP addresses instead of localhost or 127.0.0.1.