Kafka Connect and Debezium CDC Streaming
Kafka Connect Debezium enables real-time Change Data Capture (CDC) streaming from databases to Apache Kafka. Instead of building custom data synchronization code, Debezium monitors database transaction logs and streams every insert, update, and delete as an event to Kafka topics. This powers real-time analytics, search index updates, cache invalidation, and cross-service data replication without modifying application code. Because the capture happens at the log level rather than through application hooks, the source application stays completely unaware that its changes are being mirrored elsewhere.
This guide covers building production CDC pipelines with Kafka Connect and Debezium, including connector configuration for PostgreSQL and MySQL, Single Message Transforms (SMTs), schema evolution handling, and monitoring strategies. Throughout, the emphasis is on the operational details that separate a working demo from a pipeline you can trust to feed downstream systems at three in the morning without manual babysitting.
CDC Architecture Overview
Change Data Capture Pipeline
┌──────────────┐ Transaction ┌──────────────┐
│ PostgreSQL │───── Log ────────▶│ Debezium │
│ (Source DB) │ (WAL/Binlog) │ Connector │
└──────────────┘ └──────┬───────┘
│
┌──────▼───────┐
│ Kafka │
│ Topics │
└──────┬───────┘
│
┌─────────────────────┼─────────────────────┐
│ │ │
┌──────▼──────┐ ┌──────▼──────┐ ┌──────▼──────┐
│ Elasticsearch│ │ Data │ │ Analytics │
│ (Search) │ │ Warehouse │ │ Service │
└─────────────┘ └─────────────┘ └────────────┘
Setting Up Kafka Connect
# docker-compose.yml — Kafka Connect with Debezium
services:
kafka:
image: confluentinc/cp-kafka:7.6.0
environment:
KAFKA_NODE_ID: 1
KAFKA_PROCESS_ROLES: broker,controller
KAFKA_CONTROLLER_QUORUM_VOTERS: '1@kafka:9093'
KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
CLUSTER_ID: 'q1Sh-9_ISia_zwGINzRvyQ'
connect:
image: debezium/connect:2.6
ports:
- "8083:8083"
environment:
BOOTSTRAP_SERVERS: kafka:9092
GROUP_ID: connect-cluster
CONFIG_STORAGE_TOPIC: connect-configs
OFFSET_STORAGE_TOPIC: connect-offsets
STATUS_STORAGE_TOPIC: connect-status
KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter
VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter
depends_on:
- kafka
postgres:
image: postgres:16
environment:
POSTGRES_DB: myapp
POSTGRES_USER: appuser
POSTGRES_PASSWORD: secret
command:
- postgres
- -c
- wal_level=logical # Required for CDC!Configuring the Debezium Connector
The connector configuration specifies which database and tables to monitor. Moreover, Debezium supports fine-grained control over which changes to capture:
{
"name": "postgres-cdc-connector",
"config": {
"connector.class": "io.debezium.connector.postgresql.PostgresConnector",
"database.hostname": "postgres",
"database.port": "5432",
"database.user": "debezium",
"database.password": "${file:/secrets/db-password}",
"database.dbname": "myapp",
"topic.prefix": "myapp.cdc",
"plugin.name": "pgoutput",
"slot.name": "debezium_slot",
"publication.name": "debezium_pub",
"table.include.list": "public.orders,public.customers,public.products",
"column.exclude.list": "public.customers.ssn,public.customers.password_hash",
"key.converter": "org.apache.kafka.connect.json.JsonConverter",
"key.converter.schemas.enable": false,
"value.converter": "org.apache.kafka.connect.json.JsonConverter",
"value.converter.schemas.enable": false,
"transforms": "route,unwrap,timestamp",
"transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
"transforms.route.regex": "myapp\\.cdc\\.public\\.(.*)",
"transforms.route.replacement": "cdc.$1",
"transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
"transforms.unwrap.drop.tombstones": false,
"transforms.unwrap.delete.handling.mode": "rewrite",
"transforms.unwrap.add.fields": "op,source.ts_ms",
"transforms.timestamp.type": "org.apache.kafka.connect.transforms.TimestampConverter$Value",
"transforms.timestamp.target.type": "string",
"transforms.timestamp.field": "__source_ts_ms",
"transforms.timestamp.format": "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'",
"heartbeat.interval.ms": 10000,
"snapshot.mode": "initial",
"tombstones.on.delete": true,
"decimal.handling.mode": "string",
"errors.tolerance": "all",
"errors.deadletterqueue.topic.name": "cdc.dlq",
"errors.deadletterqueue.context.headers.enable": true
}
}# Deploy the connector
curl -X POST http://localhost:8083/connectors \
-H "Content-Type: application/json" \
-d @postgres-cdc-connector.json
# Check connector status
curl http://localhost:8083/connectors/postgres-cdc-connector/status | jq
# List topics created
kafka-topics --bootstrap-server kafka:9092 --list | grep cdcSnapshots, Replication Slots, and the Initial Load
When a connector starts for the first time, it performs an initial snapshot, reading the entire contents of the included tables before switching to streaming mode. The snapshot.mode setting controls this behavior, and choosing the wrong value is a common source of confusion. The default initial mode snapshots once, while never skips the snapshot entirely and begins streaming from the current log position. For very large tables, the documentation recommends the incremental snapshot feature, which reads the table in chunks and interleaves snapshot rows with live changes rather than holding a long-running read.
On PostgreSQL, the connector creates a logical replication slot that pins the write-ahead log until Debezium has consumed it. This is the single most important operational fact to internalize. If a connector is paused, deleted, or simply falls behind, the slot prevents PostgreSQL from recycling WAL segments, and disk usage climbs steadily until the volume fills. Therefore, teams typically alert on pg_replication_slots.confirmed_flush_lsn lag and always drop abandoned slots with SELECT pg_drop_replication_slot('debezium_slot') after retiring a pipeline.
Single Message Transforms in Depth
The raw Debezium envelope is verbose. Each event wraps the changed row inside before, after, source, and op fields, which is rarely what a downstream consumer wants. The ExtractNewRecordState SMT (commonly called “unwrap”) flattens this into the new row state with a few metadata columns appended. As a result, an UPDATE arrives looking almost like the table row itself, plus an __op code of u and a source timestamp.
SMTs run in the order declared in the transforms property, and order matters. In the configuration above, route rewrites the topic name first, then unwrap flattens the payload, then timestamp converts the epoch milliseconds into an ISO-8601 string. A frequent mistake is placing a field-manipulation SMT before unwrap, where the target field does not yet exist at the top level. When transformation logic grows beyond a few SMTs, teams typically move it out of Connect and into a dedicated stream processor such as Kafka Streams or Flink, since chaining many SMTs becomes hard to reason about and debug.
Consuming CDC Events
CDC events contain the full before and after state of each record. Therefore, consumers can implement any downstream processing logic:
@Component
public class OrderCdcConsumer {
private final ElasticsearchClient esClient;
private final CacheManager cacheManager;
@KafkaListener(topics = "cdc.orders",
groupId = "search-indexer")
public void handleOrderChange(ConsumerRecord<String, String> record) {
ObjectMapper mapper = new ObjectMapper();
JsonNode event = mapper.readTree(record.value());
String operation = event.get("__op").asText();
String orderId = event.get("id").asText();
switch (operation) {
case "c", "u" -> {
// Create or update: index in Elasticsearch
OrderDocument doc = mapper.treeToValue(
event, OrderDocument.class);
esClient.index(i -> i
.index("orders")
.id(orderId)
.document(doc));
// Invalidate cache
cacheManager.getCache("orders")
.evict(orderId);
}
case "d" -> {
// Delete: remove from search index
esClient.delete(d -> d
.index("orders")
.id(orderId));
cacheManager.getCache("orders")
.evict(orderId);
}
}
}
}
Idempotency and Exactly-Once Semantics
CDC consumers must be idempotent, because at-least-once delivery is the realistic default. A connector restart, a rebalance, or a network blip can cause an event to be redelivered, and a non-idempotent consumer would double-apply the change. The example above happens to be idempotent already: indexing a document by its primary key in Elasticsearch overwrites any prior version, so reprocessing the same event produces the same end state.
When downstream side effects are not naturally idempotent, such as incrementing a counter or sending an email, consumers typically deduplicate on a stable key. Debezium events expose the source log sequence number through the source block, and many teams store the last-applied LSN per key to reject stale or duplicate events. Ordering is another subtlety: Kafka guarantees order only within a partition, so to preserve per-row ordering you must key the topic by the primary key. The default Debezium key already does this, but a misconfigured route or a custom key SMT can silently break the guarantee.
Schema Evolution and Compatibility
Databases change. Columns get added, types widen, and tables get renamed, and a CDC pipeline must survive all of it without a 3 a.m. page. When using JSON converters as shown here, schema changes flow through transparently because JSON is schemaless on the wire, but downstream consumers must tolerate unexpected or missing fields. In contrast, teams running the Avro converter with a Schema Registry get strong compatibility enforcement: a backward-compatible change such as adding a nullable column is accepted, while an incompatible change such as dropping a required field is rejected at the registry before it can corrupt consumers.
For PostgreSQL specifically, DDL changes are not captured by the logical decoding stream the way they are on MySQL’s binlog. Consequently, adding a column appears simply as new fields showing up in subsequent events, while a column rename can look like a drop plus an add. A common pattern is to treat the database schema as a contract, version downstream consumers defensively, and roll out additive changes first so that producers and consumers never assume a field exists before it is guaranteed to.
Monitoring and Operations
# Key JMX metrics to monitor
# Source connector metrics:
# - MilliSecondsBehindSource: replication lag
# - TotalNumberOfEventsSeen: throughput
# - NumberOfEventsFiltered: filtered events
# - SnapshotCompleted: initial snapshot status
# Kafka Connect metrics:
# - connector-status: running/paused/failed
# - task-count: number of active tasks
# - offset-commit-success-rate: offset reliability
# Alert thresholds:
# MilliSecondsBehindSource > 30000 → WARNING
# MilliSecondsBehindSource > 300000 → CRITICAL
# connector-status != RUNNING → CRITICALBeyond raw metrics, the dead letter queue configured earlier deserves an alarm of its own. Because errors.tolerance is set to all, a malformed event will not crash the task; it will instead be routed to the cdc.dlq topic with the failure reason captured in the message headers. That is exactly the behavior you want for resilience, but a silent DLQ that nobody watches is just data loss with extra steps. Therefore, teams typically alert when the DLQ partition lag grows and periodically inspect the failure headers to distinguish a genuine bug from a transient downstream outage.
When NOT to Use Kafka Connect with Debezium
CDC adds operational complexity with replication slots, connector management, and schema evolution handling. If your data synchronization needs are simple and infrequent, batch ETL jobs may be more appropriate. Furthermore, databases with extremely high write throughput (100,000+ transactions/second) can generate more CDC events than Kafka can handle without significant infrastructure investment. Additionally, if you only need to replicate data between two PostgreSQL instances, native logical replication is simpler than the Debezium pipeline. Evaluate whether the real-time requirement justifies the operational overhead.
There are also failure modes worth weighing before committing. A replication slot tied to a connector that nobody is maintaining can quietly fill a primary database’s disk, which is a far worse outage than a stale cache. Likewise, the initial snapshot of a multi-terabyte table can run for hours and add read load to a production primary, so teams often point the snapshot at a read replica or use incremental snapshots. If your team lacks the appetite to operate Kafka, Connect, and the schema tooling together, a managed CDC offering or a simpler polling-based sync may deliver more value for less risk.
Key Takeaways
- Kafka Connect Debezium captures database changes from transaction logs without modifying application code
- Single Message Transforms (SMTs) reshape events — route topics, extract new state, convert timestamps
- Replication slots pin WAL until consumed — abandoned slots can fill a primary database’s disk
- Make consumers idempotent and key topics by primary key to preserve per-row ordering
- Dead letter queues prevent poison messages from blocking the entire pipeline
- Monitor MilliSecondsBehindSource to detect replication lag before it impacts downstream consumers
- Use CDC for real-time search indexing, cache invalidation, analytics, and cross-service data replication
Related Reading
- Redis Streams Event-Driven Architecture
- PostgreSQL 18 New Features and Performance
- ClickHouse vs DuckDB Analytical Databases
- Event-Driven Architecture with Kafka Patterns
- Outbox Pattern for Reliable Event Publishing
External Resources
In conclusion, Kafka Connect Debezium is an essential building block for modern, event-driven data architectures. By applying the patterns and practices covered in this guide — careful snapshot planning, disciplined SMT ordering, idempotent consumers, and vigilant monitoring of replication slots and lag — you can build CDC pipelines that are robust, scalable, and maintainable. Start with the fundamentals, iterate on your implementation, and continuously measure results to ensure you are getting the most value from these approaches.