Kafka Connect CDC: PostgreSQL to Elasticsearch and JDBC

Note: when the PostgreSQL database grows to hundreds of gigabytes and search index freshness requirements drop to seconds, manual synchronization stops working. We configure Kafka Connect for streaming data replication (CDC) — it is more reliable and faster than custom poller services. For example, a marketplace with a catalog in PostgreSQL and search in Elasticsearch faces index update delays of up to 15 minutes. With our solution, the delay drops to 2 seconds (99.8% improvement), and when a node fails, data is not lost — tasks are redistributed automatically. One typical case: PostgreSQL → Debezium → Kafka → Elasticsearch. Without Kafka Connect, engineers spend weeks writing a WAL handler and dealing with duplicates. We do it in 4 days turnkey, with a fault-tolerant cluster and monitoring. Debezium documentation confirms that CDC provides streaming with minimal latency. Our solution saves up to $15,000 per year on maintenance costs compared to custom polling.

Why Kafka Connect is better than custom integration?

Let's compare the approaches:

Result: Kafka Connect provides a ready-made framework with guaranteed delivery (exactly-once with idempotent producers) and saves 80% development time and 50% operations time.

How CDC works on PostgreSQL with Debezium?

Change Data Capture (CDC) intercepts every change in the database and streams it as events. Debezium connects to PostgreSQL WAL (Write-Ahead Log) and sends INSERT/UPDATE/DELETE to Kafka. This eliminates the need for custom triggers or polling tables on a schedule. Debezium supports snapshot.mode=initial for initial load and incremental to avoid locks on large tables. After configuration, every change appears in a Kafka topic within milliseconds. Average lag is 100 ms, throughput up to 10,000 messages/sec. We've processed over 500 million events for clients with 99.99% data consistency.

Turnkey Kafka Connect setup: 4 steps

Follow these steps to set up your streaming database integration. Step 1: Prepare PostgreSQL and Kafka

Enable logical replication: wal_level = logical, create a publication for the required tables and a debezium user with SELECT privileges. On the Kafka side, check bootstrap.servers, retention settings, and compact topics for Debezium.

Step 2: Deploy Kafka Connect in distributed mode

A cluster of 2–3 nodes with internal topics for configuration. The configuration is fully typed (see below). We use Avro with Schema Registry — this guarantees schema compatibility when the database changes.

Step 3: Configure Debezium Source Connector

Debezium reads the PostgreSQL WAL and sends every change to Kafka. Configure snapshot.mode=initial, transforms for extracting the new row, and tombstone for DELETE. For large tables (billions of rows), use incremental snapshot to avoid blocking the database.

Step 4: Sink connectors to Elasticsearch and PostgreSQL

For the search index — Elasticsearch Sink with batching of 500 records and retry backoff. For the analytics database — JDBC Sink with upsert and pk.mode=record_key. At each stage, test INSERT/UPDATE/DELETE and lag.

How to deploy Kafka Connect and connectors?

Below are the key configurations for launching a distributed cluster and typical connectors.

# distributed properties (connect-distributed.properties) bootstrap.servers=kafka-1:9092,kafka-2:9092,kafka-3:9092 group.id=kafka-connect-cluster config.storage.topic=connect-configs offset.storage.topic=connect-offsets status.storage.topic=connect-statuses config.storage.replication.factor=3 offset.storage.replication.factor=3 status.storage.replication.factor=3 offset.flush.interval.ms=10000 rest.host.name=0.0.0.0 rest.port=8083 rest.advertised.host.name=connect-1.internal rest.advertised.port=8083 plugin.path=/opt/kafka/plugins 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 

Before running Debezium, configure PostgreSQL:

ALTER SYSTEM SET wal_level = logical; ALTER SYSTEM SET max_replication_slots = 10; ALTER SYSTEM SET max_wal_senders = 10; CREATE USER debezium WITH REPLICATION LOGIN PASSWORD 'secure_password'; GRANT CONNECT ON DATABASE myapp TO debezium; GRANT USAGE ON SCHEMA public TO debezium; GRANT SELECT ON ALL TABLES IN SCHEMA public TO debezium; ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO debezium; CREATE PUBLICATION debezium_pub FOR TABLE products, orders, users, categories; 

Now register the connectors via REST API. Here is an example of Debezium Source and JDBC Sink in one block:

# Debezium Source curl -X POST http://connect-1:8083/connectors -H "Content-Type: application/json" -d '{ "name": "postgres-source-connector", "config": { "connector.class": "io.debezium.connector.postgresql.PostgresConnector", "database.hostname": "postgres.internal", "database.port": "5432", "database.user": "debezium", "database.password": "secure_password", "database.dbname": "myapp", "database.server.name": "myapp-pg", "topic.prefix": "myapp", "table.include.list": "public.products,public.orders,public.users", "plugin.name": "pgoutput", "publication.name": "debezium_pub", "slot.name": "debezium_slot", "snapshot.mode": "initial", "snapshot.isolation.mode": "read_committed", "decimal.handling.mode": "double", "time.precision.mode": "connect", "tombstones.on.delete": "true", "heartbeat.interval.ms": "10000", "transforms": "unwrap", "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState", "transforms.unwrap.delete.handling.mode": "rewrite", "transforms.unwrap.add.fields": "op,ts_ms,source.ts_ms" } }' # JDBC Sink curl -X POST http://connect-1:8083/connectors -H "Content-Type: application/json" -d '{ "name": "postgres-sink-connector", "config": { "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector", "tasks.max": "4", "topics": "myapp.analytics.events", "connection.url": "jdbc:postgresql://analytics-pg:5432/analytics", "connection.user": "kafka_writer", "connection.password": "secure_password", "auto.create": "false", "auto.evolve": "false", "insert.mode": "upsert", "pk.mode": "record_key", "pk.fields": "id", "table.name.format": "analytics.${topic}", "batch.size": "1000", "db.timezone": "UTC", "transforms": "dropPrefix", "transforms.dropPrefix.type": "org.apache.kafka.connect.transforms.ReplaceField$Value", "transforms.dropPrefix.exclude": "__deleted,__op,__ts_ms" } }' 

Connector management is done via REST API: status retrieval, pause, restart of failed tasks. We integrate Kafka Connect with Prometheus monitoring for real-time metrics. Prometheus JMX metrics are configured using JMX Exporter.

Typical problems and their solutions

What's included in the work

• Analysis of the current database schema and loads, selection of connectors and transformations
• PostgreSQL configuration for logical replication (WAL, publication, users)
• Installation of Kafka Connect in distributed mode on 2–3 nodes with Schema Registry
• Deployment of Debezium Source Connector with initial snapshot
• Configuration of one or more Sink connectors (Elasticsearch, JDBC, S3)
• Writing Single Message Transforms (SMT) for schema alignment
• Monitoring integration: Prometheus JMX Exporter, Grafana dashboard, Slack alerts
• Documentation of topic schema and configurations
• Team training (2 hours: basic operations, restart, diagnostics)

Timeline

Over 10 years we have delivered 50+ integration pipelines on PostgreSQL, MySQL, and MongoDB. Get a consultation for your project – our engineers will analyze your schema and load and propose the optimal architecture. Contact us — we will assess your project for free in 2 hours. Order Kafka Connect setup — and we guarantee change delivery in seconds.