Building a CDC Pipeline with Debezium for Real-Time Search Indexing

Modern search architectures require sub-second data parity between transactional databases and query engines. Implementing a Change Data Capture (CDC) Setup eliminates polling latency, reduces primary database load on your Data Ingestion & Synchronization Pipelines, and enables deterministic event ordering. This guide details the exact Debezium configuration, failure diagnostics, and index synchronization patterns required for production environments. The connector wiring here is engine-agnostic at the source layer; for relational-specific binlog wiring see the CDC connector setup for MySQL, and for document stores see the CDC connector setup for MongoDB.

Prerequisites & Infrastructure Alignment

Ensure your Data Ingestion & Synchronization Pipelines infrastructure meets baseline throughput and durability requirements. Deploy Kafka Connect in distributed mode with a minimum of three worker nodes for high availability. Verify network ACLs allow source database binlog/WAL access. Confirm outbound connectivity to the search cluster and Schema Registry.

Validate PostgreSQL logical decoding prerequisites before deployment. Set wal_level=logical, max_replication_slots=10, and max_wal_senders=10 in postgresql.conf. Restart the database instance to apply changes. Create a dedicated replication role with REPLICATION and LOGIN privileges. Grant SELECT access to all target tables.

Exact Debezium Connector Configuration

Deploy the connector via the Kafka Connect REST API. Use snapshot.mode=initial for bootstrapping historical data. Transition seamlessly to streaming once the snapshot completes. Enable heartbeat.interval.ms=5000 to maintain WAL retention on idle tables. Configure Single Message Transforms (SMTs) for payload flattening. Route events to search-specific Kafka topics. Avoid custom serialization unless strictly required. Stick to Avro or Protobuf with Schema Registry enforcement.

curl -X POST http://kafka-connect:8083/connectors \
 -H "Content-Type: application/json" \
 -d '{
 "name": "postgres-cdc-search-sync",
 "config": {
 "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
 "database.hostname": "prod-db.internal",
 "database.port": "5432",
 "database.user": "cdc_reader",
 "database.password": "${DB_PASS}",
 "database.dbname": "app_db",
 "topic.prefix": "search.cdc",
 "schema.history.internal.kafka.topic": "schema-changes.search",
 "schema.history.internal.kafka.bootstrap.servers": "kafka-broker-1:9092,kafka-broker-2:9092,kafka-broker-3:9092",
 "snapshot.mode": "initial",
 "heartbeat.interval.ms": "5000",
 "transforms": "unwrap,route",
 "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
 "transforms.unwrap.add.fields": "op,source.ts_ms",
 "transforms.unwrap.delete.handling.mode": "rewrite",
 "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
 "transforms.route.regex": "([^.]+)\\.([^.]+)\\.([^.]+)",
 "transforms.route.replacement": "search-index.$3",
 "key.converter": "io.confluent.connect.avro.AvroConverter",
 "value.converter": "io.confluent.connect.avro.AvroConverter",
 "key.converter.schema.registry.url": "http://schema-registry:8081",
 "value.converter.schema.registry.url": "http://schema-registry:8081"
 }
 }'

Target Elasticsearch or OpenSearch using idempotent upserts. Derive document IDs directly from primary keys. Configure sink batch settings strictly. Use max.in.flight.requests=1 and batch.size=1000. Set flush.timeout.ms=30000 to prevent timeout cascades.

Diagnostic Steps for Pipeline Failures

Isolate failure domains using a structured diagnostic workflow. First, verify connector state via GET /connectors/{name}/status. Second, inspect Kafka Connect worker logs for OffsetCommit errors. Look for SerializationException stack traces. Third, run kafka-consumer-groups.sh --describe to identify lagging partitions. Fourth, validate the source database WAL retention settings. Ensure Debezium has not fallen behind the retention window. Finally, check the Schema Registry for compatibility violations on DDL changes.

# Step 1: Verify connector status
curl -s http://kafka-connect:8083/connectors/postgres-cdc-search-sync/status | jq '.tasks[].state'

# Step 2: Grep worker logs for critical errors
grep -E "ERROR|WARN" /var/log/kafka/connect.log | grep -iE "Debezium|Kafka|Schema"

# Step 3: Check consumer group lag
kafka-consumer-groups.sh --bootstrap-server kafka-broker-1:9092 \
 --group connect-postgres-cdc-search-sync --describe

# Step 4: Validate WAL retention window (PostgreSQL 13+: wal_keep_size in MB; older: wal_keep_segments in segments)
psql -U cdc_reader -d app_db -c "SHOW wal_keep_size;"  # PostgreSQL >= 13
# psql -U cdc_reader -d app_db -c "SHOW wal_keep_segments;"  # PostgreSQL < 13

Resolution Paths for Schema Drift & Index Conflicts

When upstream DDL breaks the pipeline, route malformed records to a dedicated Dead Letter Queue (DLQ) topic. Apply transforms=io.debezium.transforms.ByLogicalTableRouter for tenant isolation. Rebuild search mappings using dynamic templates to handle new fields gracefully. Implement optimistic concurrency control via _version or if_seq_no in the sink. This prevents race conditions during concurrent updates.

{
 "name": "postgres-cdc-search-sync",
 "config": {
 "errors.tolerance": "all",
 "errors.deadletterqueue.topic.name": "dlq.search.cdc",
 "errors.deadletterqueue.context.headers.enable": "true",
 "errors.log.enable": "true",
 "errors.log.include.messages": "true"
 }
}

Schema mismatch requires immediate registry alignment. Register the new schema version in the registry. Restart the connector with snapshot.mode=when_needed if historical backfill is required. Index desync demands targeted re-indexing. Enable sink-side versioning to enforce strict ordering. Implement a replay consumer for the DLQ topic to recover corrupted records safely.

Production Hardening & Observability

Instrument the pipeline with JMX exporters. Scrape metrics via Prometheus for continuous visibility. Monitor source-record-poll-rate and source-record-active-count. Track offset-commit-failure-rate closely. Set alert thresholds for consumer lag exceeding 5 seconds. Trigger alerts when error rates surpass 0.1%. Use errors.tolerance=all paired with DLQ routing. This prevents connector crashes on transient data anomalies. Schedule automated connector restarts with exponential backoff. This handles transient network partitions gracefully.

# prometheus-jmx-exporter config snippet
rules:
 - pattern: "kafka.connect<type=connect-metrics, client-id=.*><>(source-record-poll-rate)"
 name: kafka_connect_source_record_poll_rate
 type: GAUGE
 - pattern: "kafka.connect<type=connect-metrics, client-id=.*><>(offset-commit-failure-rate)"
 name: kafka_connect_offset_commit_failure_rate
 type: GAUGE

Configure Kubernetes CronJobs or systemd timers for automated recovery. Implement exponential backoff starting at 30 seconds. Cap retries at 10 attempts per failure cycle. Maintain a runbook for manual offset resets. Store offsets in Kafka internal topics for rapid state recovery.

Operating the connector after day one

Setting a connector up takes an afternoon; keeping it healthy is the actual commitment, and the failure modes that show up in month three are not the ones the quickstart prepares you for. The three that account for most incidents are unexpected re-snapshots, task failures hidden behind a healthy connector state, and topic growth that nobody sized for.

A Debezium connector is a stateful service, and the state lives in three places that must stay consistent: the source position (slot or binlog offset), the Connect offset topic, and the sink’s own idea of what it has applied. Most incidents in the months after setup are a divergence between two of those three.

The three positions a change capture pipeline must keep consistent The source retention position, the connector offset topic and the sink's applied position form a triangle; recovery means restoring agreement between them. source position slot / binlog connect offsets where it resumes sink applied what search holds deleting a connector clears the middle box but not the left one which is why orphaned slots and unexpected re-snapshots happen
Three positions, two arrows. Nearly every post-setup incident is one of these three drifting away from the other two.

The most common surprise is an unexpected re-snapshot. Deleting and recreating a connector with the same name usually reuses the stored offsets — but if the offset topic was cleared, or the connector name changed by a character, Connect finds no offsets and starts a fresh snapshot of every captured table. On a large table that means hours of load on the source and a flood of op: r events at the sink. Before recreating any connector, check whether offsets exist for that exact name.

# What position does Connect think this connector is at?
kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic connect-offsets --from-beginning --property print.key=true \
  | grep 'products-cdc' | tail -2
# An empty result means the next start WILL re-snapshot.

The second recurring issue is connector task failure without connector failure. A connector reports RUNNING while its task has failed, so a naive health check on connector state alone is green while nothing is being captured. Always check task state, and alert on it.

curl -s localhost:8083/connectors/products-cdc/status \
  | jq -r '.connector.state as $c | .tasks[].state | "connector=\($c) task=\(.)"'
# => connector=RUNNING task=FAILED     <- the state that silently stops ingestion

The third is topic growth. Change events are verbose — Debezium’s envelope carries before, after, source metadata and a schema block — so the topic can be several times the size of the underlying table churn. Enable log compaction on keyed topics where only the latest state per key matters, and set retention deliberately rather than inheriting a search cluster default that was chosen for something else.

Finally, treat the connector configuration as code. It is a JSON document that decides what your search index contains, and editing it through the REST API by hand means the running configuration and the repository diverge within weeks. Store it, apply it from CI, and diff the live configuration against the stored one on a schedule.

Health signals that must all be checked, not just connector state Connector state, task state, source lag and sink lag form four independent health signals, and only checking all four detects every stall. connector state necessary, not enough task state where failures hide source lag retained log size sink lag user-visible freshness A pipeline can be RUNNING, task-FAILED, log-accumulating and stale all at once — three of four signals green.
Four independent signals. Monitoring only the leftmost one is the most common gap in a freshly deployed capture pipeline.
Connector configuration applied from version control Configuration in the repository is applied by continuous integration and periodically diffed against the live connector so manual edits are detected. connector.json in git applied by CI live connector scheduled diff — a manual edit shows up as drift within a day
Treating the connector config as code closes the gap between what the repository claims is running and what actually is.

One last operational habit: keep a written recovery procedure for the two scenarios that actually happen — the connector has fallen behind retention, and the connector is running but its task has failed. Both are recoverable, both are time-sensitive, and neither is intuitive under pressure. A ten-line runbook naming the exact commands to check position, restart a failed task, and trigger a controlled re-snapshot turns a two-hour incident into a fifteen-minute one, and it is the cheapest artefact in this entire pipeline to produce.

Sizing and cost notes

  • Event size is typically 3–8× the row’s own size once the envelope, schema and source metadata are included. Budget topic storage from measured events, not from table size.
  • Snapshot duration scales with rows and mapping complexity; expect roughly 15–40 minutes per 10 million rows on commodity hardware, during which the source carries an extra sequential read load.
  • Task parallelism for most source connectors is one task per database, so scaling capture means splitting the capture set across connectors rather than raising tasks.max.
  • Restart cost is dominated by the replay from the last committed offset. With a 5-second commit interval and a 12,000 per second stream, a restart replays around 60,000 events — harmless with version guards, expensive without them.
  • Schema history topic must never be deleted for a MySQL connector; losing it forces a re-snapshot even though the binlog position is still valid.