Context: Why Dual-Write and Cron Jobs Will Haunt You
Imagine running an e-commerce platform processing 2,000 orders every minute. PostgreSQL serves as the source of truth, recording each transaction with full ACID compliance. Meanwhile, the product team relies on Elasticsearch to provide customers with fast product search and order history lookups at sub-50ms latencies.
The first solution teams typically consider is a dual-write pattern: the application writes to Postgres and immediately makes an API call to Elasticsearch. It sounds straightforward. However, a single 3-second Elasticsearch timeout or an intermittent network hiccup is all it takes for the two datastores to drift out of sync. You are then left writing complex retry logic and distributed transaction compensations.
The second option is scheduling a cron job every 5 minutes. Once tables grow to tens of millions of rows, polling with SELECT ... WHERE updated_at > ... exhausts database IOPS. On top of that, search data is perpetually stale by at least 5 minutes.
This is where Change Data Capture (CDC) comes to the rescue. Instead of running heavy polling queries, CDC attaches a reader directly to PostgreSQL’s Write-Ahead Log (WAL). Every row insertion, update, and deletion (INSERT, UPDATE, DELETE) is captured immediately with practically zero CPU overhead.
The combination of Debezium, Kafka, and Kafka Connect orchestrates this workflow seamlessly: Debezium streams WAL changes into events, Kafka delivers them reliably at scale, and Kafka Connect sinks the data into Elasticsearch with latencies under a few hundred milliseconds.
Spinning Up the Environment with Docker Compose
You don’t need to manually install each component on your machine. Docker Compose lets you stand up the entire cluster in minutes.
1. Prepare the docker-compose.yml File
Create a working directory and save the following docker-compose.yml file:
version: '3.8'
services:
postgres:
image: quay.io/debezium/postgres:15-alpine
ports:
- "5432:5432"
environment:
- POSTGRES_USER=postgres
- POSTGRES_PASSWORD=postgres
- POSTGRES_DB=inventory
zookeeper:
image: confluentinc/cp-zookeeper:7.5.0
environment:
ZOOKEEPER_CLIENT_PORT: 2181
ZOOKEEPER_TICK_TIME: 2000
kafka:
image: confluentinc/cp-kafka:7.5.0
depends_on:
- zookeeper
ports:
- "9092:9092"
environment:
KAFKA_BROKER_ID: 1
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
kafka-connect:
image: debezium/connect:2.4
depends_on:
- kafka
- postgres
ports:
- "8083:8083"
environment:
BOOTSTRAP_SERVERS: kafka:9092
GROUP_ID: 1
CONFIG_STORAGE_TOPIC: my_connect_configs
OFFSET_STORAGE_TOPIC: my_connect_offsets
STATUS_STORAGE_TOPIC: my_connect_statuses
elasticsearch:
image: docker.elastic.co/elasticsearch/elasticsearch:8.10.2
ports:
- "9200:9200"
environment:
- discovery.type=single-node
- xpack.security.enabled=false2. Start the Cluster
Run the services in the background:
docker compose up -dWait roughly 30 to 45 seconds for Kafka Connect to load its plugins and expose its REST API on port 8083.
Configuring the CDC Pipeline
For Debezium to read transaction logs, PostgreSQL requires wal_level = logical. The quay.io/debezium/postgres image enables this by default. If you are using an existing standalone Postgres instance or Amazon RDS, make sure this parameter is properly configured.
1. Seed Initial Data in PostgreSQL
Create a customers table and insert two sample rows:
docker compose exec -it postgres psql -U postgres -d inventory -c "
CREATE TABLE customers (
id SERIAL PRIMARY KEY,
first_name VARCHAR(255),
last_name VARCHAR(255),
email VARCHAR(255) UNIQUE,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
INSERT INTO customers (first_name, last_name, email) VALUES
('Nguyen', 'An', '[email protected]'),
('Tran', 'Binh', '[email protected]');
"2. Register the Debezium Postgres Source Connector
Submit the connector configuration to Kafka Connect’s REST API on port 8083:
curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" \
http://localhost:8083/connectors/ -d '{
"name": "postgres-source-connector",
"config": {
"connector.class": "io.debezium.connector.postgresql.PostgresConnector",
"tasks.max": "1",
"plugin.name": "pgoutput",
"database.hostname": "postgres",
"database.port": "5432",
"database.user": "postgres",
"database.password": "postgres",
"database.dbname": "inventory",
"topic.prefix": "cdc",
"table.include.list": "public.customers"
}
}'Once registered, Debezium performs an initial snapshot. Any subsequent changes will automatically stream to the Kafka topic named cdc.public.customers.
3. Sink Data to Elasticsearch with Unwrap SMT
By default, Debezium wraps change events in a complex envelope containing schemas, along with before and after states. Elasticsearch only requires the latest flattened payload. We use the ExtractNewRecordState Single Message Transform (Unwrap SMT) to extract the clean document payload before indexing:
curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" \
http://localhost:8083/connectors/ -d '{
"name": "elastic-sink-connector",
"config": {
"connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
"tasks.max": "1",
"topics": "cdc.public.customers",
"connection.url": "http://elasticsearch:9200",
"type.name": "_doc",
"key.ignore": "false",
"schema.ignore": "true",
"transforms": "unwrap",
"transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
"transforms.unwrap.drop.tombstones": "false"
}
}'If your stack does not use Confluent’s sink connector plugin, you can replace it with a lightweight Python or Go worker that consumes messages from Kafka and pushes them to Elasticsearch via the _bulk API.
Testing & Production Operations
The pipeline is fully connected. Now it’s time to verify end-to-end latency.
1. Check Connector Status
Confirm that both connectors are running smoothly:
curl -s http://localhost:8083/connectors/postgres-source-connector/status | jq .
curl -s http://localhost:8083/connectors/elastic-sink-connector/status | jq .Verify that the state field displays RUNNING. If it indicates FAILED, inspect the container logs using docker compose logs kafka-connect to check for authentication errors or missing plugins.
2. Verify Real-Time Data Synchronization
Insert a new customer record into Postgres:
docker compose exec -it postgres psql -U postgres -d inventory -c "
INSERT INTO customers (first_name, last_name, email) VALUES ('Le', 'Cuong', '[email protected]');
"Almost instantaneously (typical latencies range between 150ms and 300ms), query Elasticsearch directly:
curl -s http://localhost:9200/cdc.public.customers/_search?q=last_name:Cuong | jq .hits.hitsThe record for customer ‘Le Cuong’ appears in the search index immediately, without touching a single line of sync code in your application.
3. Three Vital Metrics for Production
Running CDC in production requires careful monitoring of the following key failure points:
- Postgres Replication Slot Lag: This is the most common trap. If Kafka Connect fails or stops unexpectedly, PostgreSQL retains all incoming WAL files so it can resume replication later. Disk usage can quickly balloon from 20GB to 200GB within hours, bringing your primary database down. Always set up alerts to monitor WAL retention across replication slots using
pg_replication_slots. - Kafka Consumer Lag: Measures the gap between the latest produced offset and the offset processed by the Elasticsearch sink. Steadily increasing lag indicates a write bottleneck in Elasticsearch. Mitigate this by allocating additional sink tasks or tuning Elasticsearch’s index refresh interval.
- Kafka Connect JMX Metrics: Integrate the Prometheus JMX Exporter to track record poll rates and commit times, enabling you to proactively identify throughput bottlenecks before end users are impacted.
