Change Data Capture (CDC) Setup

Real-time search indexing requires deterministic data propagation from source databases to distributed index clusters. A properly configured CDC pipeline intercepts database transaction logs. It decodes row-level mutations and streams them to downstream consumers. This architecture avoids impacting primary OLTP workloads.

Within modern Data Ingestion & Synchronization Pipelines, CDC replaces inefficient polling mechanisms with event-driven log parsing. This shift directly addresses latency constraints that block traditional workflows. It enables sub-second index refresh cycles at scale. The path is uniform regardless of source — the MySQL connector setup and the MongoDB connector setup differ only in how each engine exposes its change stream.

The diagram below traces a single mutation from the database write-ahead log through a connector and broker to the search sink.

CDC pipeline from transaction log to search sink A write-ahead log or binlog feeds a CDC connector that publishes change events to Kafka, which a sink consumer applies to the search index. WAL / Binlog source DB CDC Connector Kafka topic / partitions Search Sink index

Transaction Log Interception & Decoding

Database engines expose transaction history through proprietary binary formats. PostgreSQL uses Write-Ahead Logs (WAL). MySQL relies on binlogs. Oracle utilizes redo logs. Row-level format is mandatory for accurate state reconstruction.

Enable binary logging on the source RDBMS with a retention window exceeding maximum expected downtime. Configure the retention period to prevent snapshot fallbacks during brief network partitions. This ensures continuous log availability for connector recovery.

# PostgreSQL: postgresql.conf
wal_level = logical
max_wal_senders = 10
max_replication_slots = 10

Deploy the CDC connector with a minimal resource footprint. Configure heartbeat intervals to maintain replication slot activity during idle periods. Select snapshot modes carefully; initial captures full state while never assumes pre-existing data. Engine-specific log access varies: row-based binlog handling for MySQL is covered in the MySQL connector setup, and oplog-based change streams for document stores in the MongoDB connector setup.

This streaming approach eliminates synchronization gaps inherent in Batch vs Streaming Ingestion workflows. CPU overhead typically remains below 3% on the primary node when parsing is offloaded to a dedicated connector host.

Schema Evolution & Data Mapping

Database schemas evolve independently from search index mappings. Type coercion must handle numeric precision shifts and timezone conversions explicitly. Nullable field transitions require explicit default values to prevent mapping rejections.

Map database schema to search index types using a dedicated transformation layer. Implement denormalization logic before data reaches the indexing queue. This reduces join overhead during query execution.

# Kafka Connect Single Message Transform (SMT) for field coercion
transforms: "flatten,coerce"
transforms.flatten.type: "org.apache.kafka.connect.transforms.Flatten$Value"
transforms.flatten.delimiter: "."
transforms.coerce.type: "org.apache.kafka.connect.transforms.Cast$Value"
transforms.coerce.spec: "price:float64,updated_at:string"

Maintain backward-compatible index mapping updates by appending new fields rather than modifying existing ones. Strict type enforcement prevents silent data corruption. Flexible mapping introduces index bloat but accelerates iteration.

While application-layer Webhook-Driven Sync Patterns provide lightweight state notifications, they introduce coupling risks. CDC decouples schema evolution from application deployment cycles.

Delivery Semantics & Backpressure Management

Exactly-once delivery guarantees require transactional outbox patterns or idempotent consumers. At-least-once semantics are simpler but mandate deduplication logic at the search sink. Configure message broker partitions to preserve causal ordering per primary key.

Implement consumer offset tracking with automated replay triggers for pipeline failures. Dead-letter queues capture malformed payloads without halting the main stream. Set consumer lag thresholds to trigger circuit breakers before index corruption occurs.

# Kafka Consumer Configuration for Idempotency & DLQ
enable.auto.commit=false
isolation.level=read_committed
max.poll.records=500
dead.letter.queue.topic=cdc-dlq-search-index

Backpressure management relies on dynamic fetch size adjustments. Monitor broker queue depth and scale consumer groups horizontally. Ensure partition count matches the maximum parallel indexing capacity.

Production Validation & Monitoring

Deploy latency and lag dashboards with precise alert thresholds. Set consumer backlog alerts at 500ms to catch degradation early. Replay testing validates recovery procedures without impacting production indexes.

Index consistency verification requires periodic checksum comparisons between source tables and search documents. Implement automated reconciliation jobs that run during low-traffic windows. Track end-to-end propagation latency across every pipeline stage.

# Prometheus Alert Rule for Consumer Lag
- alert: CDCConsumerLagHigh
 expr: kafka_consumer_group_lag > 500
 for: 2m
 labels:
 severity: warning
 annotations:
 summary: "Search CDC consumer lag exceeds 500 messages"

Target sub-second index freshness while balancing broker partition counts. Maintain zero downtime deployments with minimal mapping reindex overhead. For a production-ready reference implementation, consult Building a CDC pipeline with Debezium, which details Kafka Connect integration and search index sink configuration.

Metric Tradeoff Target
Index Freshness Sub-second latency vs increased broker partition count and network I/O < 1.5s p95 end-to-end propagation
Database Overhead Log parsing CPU cycles vs OLTP query performance degradation < 3% CPU increase on primary node
Schema Drift Handling Strict type enforcement vs flexible mapping with potential index bloat Zero downtime deployments with < 5% mapping reindex overhead

Prerequisites

  • A source database whose replication log you are permitted to read, and a maintenance window if enabling it requires a restart.
  • A replication-capable role with read access to the captured tables and nothing else.
  • A consumer runtime that can hold state — a connector framework, or a service with durable offset storage.
  • Disk headroom on the source for log retention covering your worst-case consumer downtime.
  • A version-guarded writer on the sink, because at-least-once delivery is the only guarantee you will get.

Step-by-Step Implementation

The connector-specific detail differs by database — Postgres uses replication slots, MySQL uses binlog positions, MongoDB uses change streams — but the pipeline shape below is identical across all three.

1. Decide the capture set before touching the database

Capture the tables the index needs and nothing else. A capture set defined as “everything” couples your search pipeline to every future migration in the database, and it multiplies the event volume by the ratio of total writes to writes you care about — frequently a factor of ten or more.

Verify: list the captured tables and check each one against a field actually used in the index mapping. A table nobody’s document reads is a table that should not be captured.

2. Snapshot before streaming, from a consistent point

The connector takes a consistent snapshot of the existing rows, records the log position at which the snapshot was taken, then streams forward from exactly that position. That handoff is what guarantees no gap and no duplicate window.

{
  "snapshot.mode": "initial",
  "incremental.snapshot.chunk.size": "8192",
  "signal.data.collection": "public.debezium_signal"
}

Verify: after the snapshot completes, the document count in the index should equal the source row count, and the connector’s reported position should be advancing.

curl -s 'localhost:8083/connectors/products-cdc/status' | jq -r '.connector.state, .tasks[0].state'
# => RUNNING
# => RUNNING

3. Translate the change envelope into index operations

Every CDC event carries before, after, and an operation code. The single most common ingestion bug in this area is treating all three operations as upserts, which leaves deleted rows in the index forever.

# apply.py — the three operations are genuinely different
def to_op(evt: dict) -> dict:
    p = evt["payload"]
    op = p["op"]                        # c=create u=update d=delete r=snapshot read
    if op == "d":
        return {"_op_type": "delete", "_index": "products",
                "_id": p["before"]["id"], "version": source_version(p),
                "version_type": "external_gte"}
    row = p["after"]
    return {"_op_type": "index", "_index": "products", "_id": row["id"],
            "version": source_version(p), "version_type": "external_gte",
            "_source": transform(row)}

Verify: delete a row in the source and confirm the document disappears from the index within the freshness budget.

4. Make the consumer restart-safe

Commit the consumer position only after the corresponding writes have been acknowledged by the index. Committing first turns every crash into silent data loss; committing after turns it into harmless replay, which the version guard absorbs.

# consume.py — commit AFTER the sink acknowledges, never before
for batch in stream.poll(max_records=500):
    ops = [to_op(evt) for evt in batch]
    ok, errors = bulk(es, ops, raise_on_error=False)
    if errors:
        route_failures(errors)          # dead-letter or retry, per classification
    stream.commit(batch.last_offset)    # only now is the position durable

Verify: kill the consumer mid-batch and restart it. Document counts must be unchanged, and the log should show a small number of superseded (409) writes — the replay working as designed.

5. Monitor lag as a first-class signal

Connector lag is the freshness of your index expressed as a number. Everything else in the pipeline is invisible to users; this one is not.

# Seconds between the source commit and the event reaching the sink.
max_over_time(debezium_metrics_MilliSecondsBehindSource[5m]) / 1000

Verify: the metric should sit in single-digit seconds during normal operation and recover to that band within minutes of any spike.

Snapshot to streaming handoff in a change capture pipeline A consistent snapshot is taken at a recorded log position, streaming resumes from exactly that position, and the two phases together cover every row with no gap and no duplicate window. snapshot phase reads existing rows recorded position LSN / GTID / token streaming phase resumes from that exact point every row is covered exactly once by one phase or the other a snapshot taken without recording the position is the classic source of a silent gap
The handoff, not the snapshot, is the part that has to be right. Recording the position atomically with the snapshot is what makes the guarantee hold.

A word on what change capture does not give you. It delivers row-level changes in commit order, which is not the same as delivering your domain’s changes in a useful shape. A product in the index is usually a join across three or four tables, and CDC hands you four independent streams of row changes with no notion of the entity they compose. Reassembling them is your problem, and there are only two honest approaches: enrich each event by reading the other tables at consume time (simple, adds source load, may read a newer state than the event describes), or maintain a materialised view in the source that CDC captures directly (more moving parts, but the join is done transactionally where the data lives).

Teams routinely underestimate this. The connector is configured in an afternoon and the enrichment layer takes the following two weeks, because every question about consistency — what happens when the variant row changes but the product row does not, what a delete on one side means for the composite document — has to be answered explicitly. Deciding the enrichment strategy before configuring the connector saves rebuilding the consumer once the shape becomes obvious.

Reassembling one search document from several change streams Three row-level change streams for products, variants and inventory converge on an enrichment step that produces a single composite search document. products stream variants stream inventory stream enrichment join by product id one composite document what search actually needs the connector gives you the left column; the middle box is the work
Change capture delivers rows; search needs entities. The enrichment step between them is where most of the engineering time actually goes.

Configuration Reference

Name Default Type Effect
snapshot.mode initial enum Whether to read existing rows before streaming. never skips the snapshot — correct only when another process has already populated the index.
heartbeat.interval.ms 0 (off) integer (ms) Emits a periodic marker so the consumer position advances even when captured tables are idle. Leaving it off is the main cause of unbounded log retention.
max.batch.size 2048 integer Events per poll. Larger batches amortise sink round trips but raise the replay volume after a crash.
table.include.list none list The capture set. An empty value means everything, which couples the pipeline to every table in the database.
tombstones.on.delete true boolean Emits a null-valued record after a delete so log-compacted topics drop the key. Harmless for the index, essential for topic compaction.
decimal.handling.mode precise enum precise encodes decimals as byte strings that most sinks cannot compare; double is usually what a search index actually wants.

One organisational note that matters more than any setting: change capture creates a dependency from your search team onto the database team’s migration process, and that dependency is invisible in most org charts. The database team ships a migration; the search index breaks; the search team investigates. Making the dependency explicit — a list of captured tables published where migrations are reviewed, and a named owner on the search side — converts a recurring surprise into a routine coordination step. Teams that skip this end up rediscovering the coupling once per quarter, usually during an incident.

Failure Modes & Debugging

Log retention exhausted while the consumer was down

Symptom: the connector restarts and immediately fails with an error about a position that no longer exists; the only recovery is a fresh snapshot.

Root cause: the database discarded log segments the consumer had not yet read, because retention was sized for replication rather than for a search consumer’s worst-case downtime.

Remediation: size retention against your longest plausible outage — including a weekend — and alert when a consumer’s position falls within a safety margin of the oldest retained segment. Re-snapshotting is a valid recovery, but it is hours, not minutes.

Deletes present in the log but absent from the index

Symptom: the source row count falls, the index count does not; removed records keep appearing in results.

Root cause: the consumer treats every event as an upsert, so a delete event with after: null either throws or writes an empty document.

Remediation: branch explicitly on the operation code, and add a test fixture for each of create, update, delete, and snapshot-read. This is the single most common CDC consumer bug.

Event storm after a bulk maintenance statement

Symptom: a single UPDATE on the source produces millions of change events in seconds, saturating the sink and tripping backpressure.

Root cause: log-based capture is row-level. One statement touching ten million rows is ten million events, and they all arrive after the transaction commits rather than during it.

Remediation: chunk bulk maintenance into transactions of a few thousand rows, and make sure the sink has adaptive backpressure so a storm degrades throughput instead of the pipeline.

Ordering violated by a parallel sink writer

Symptom: occasional stale field values that correct themselves on the next update to the same row.

Root cause: the consumer reads in order but writes with several concurrent workers, so two updates to one document can land out of order.

Remediation: partition sink work by a hash of the document id so one document is always handled by one worker, and carry the source position as an external version so the engine rejects a stale write regardless.

Before the numbers, one framing that helps when capacity planning: change capture converts a state problem into a flow problem. A nightly export asks “how big is the table?”; a capture pipeline asks “how fast does the table change?” Those are unrelated quantities, and a small table with heavy churn can generate far more work than a huge, static one. Size the pipeline from measured change rate, not from row count, and re-measure after any change to upstream write patterns.

Performance & Scale Notes

  • Decoding overhead on the source is typically 2–6% of CPU for a moderate write rate, and it is paid on the primary. It is rarely the constraint; log retention disk almost always bites first.
  • Event volume is row-level, not statement-level. A table with a nightly recalculation job touching every row produces one event per row per night, which for a 50-million-row table is 50 million events in a window that may be minutes long.
  • End-to-end lag in a healthy pipeline runs 2–15 seconds: sub-second to decode and publish, the rest spent batching at the sink. If lag is dominated by the sink, bulk throughput tuning moves it; if it is dominated by decoding, the capture set is too broad.
  • Consumer parallelism is bounded by partition count, and partitioning must be keyed on document id to preserve per-document ordering. Increasing partitions later requires care, because the key-to-partition mapping changes and in-flight ordering guarantees break during the transition.
  • Snapshot duration scales with row count and is the practical limit on how quickly you can recover from a lost position. For a 200-million-row table, expect hours — which is why retention sizing, not snapshot speed, is the lever that matters.

Benchmark the pipeline the way it will fail: replay a large backlog into it and measure the drain rate, rather than measuring steady-state throughput on an idle stream. The drain rate is what determines how long search is stale after an incident, and it is typically 30–50% below the steady-state number because the sink is also absorbing the burst.