Design change data capture (CDC)

It is 9:14 pm on the first night of a Big Billion Days–style sale. A seller drops a phone from ₹21,999 to ₹18,999 and an UPDATE commits in the catalogue database. Within seconds, search should rank the cheaper phone, the product-page cache should drop the old price, the fraud model should notice forty sellers doing the same, and the lakehouse should record it. Making the catalogue service write to all of those — dual writes — fails in the worst way: one write succeeds, the next times out, and nobody notices. Change data capture (CDC) turns it around: the database already records every committed change in its own log, so you read that log and publish each change as an event. Here you design a CDC platform for a large e-commerce and payments company: 200 PostgreSQL and MySQL databases, 2,000 tables, 100,000 row changes a second at peak. The hard parts are reading the log without hurting the database, merging a copy of existing rows with the live stream without gaps or stale overwrites, keeping each row’s changes in order, and surviving failovers, schema changes and consumers that fall days behind.

Understand the problem

A CDC platform is a replication system whose replicas are not databases. Its output feeds a search index, caches, a lakehouse, a fraud model and dozens of microservices, each with its own speed and its own idea of what a row means. If you need the basics first — what CDC is and why it beats dual writes — read the change data capture concept page; this page designs the platform. Two promises shape everything: every committed change gets out, in order per row, even if late; and the source database, which is taking real payments, barely notices you are there.

Functional requirements

1. Consumers should receive every committed insert, update and delete as an event with the row’s before and after images and source metadata: database, table, primary key, log position (LSN or GTID), transaction ID and commit time. Rolled-back work never appears.
2. Consumers should be able to bootstrap a table: get every existing row — an initial snapshot, a re-snapshot or a backfill of one key range — and continue with the live stream, with no gap and no stale overwrite at the seam.
3. Many independent consumers should read the same changes at their own pace, in commit order per primary key, and replay up to 7 days back.
4. Operators should be able to onboard databases and tables and ship schema changes without downtime on the source or the pipeline.

Non-functional requirements

• Scale. 200 databases (about 140 PostgreSQL, 60 MySQL), 2,000 tables, 100k row changes/s at peak on sale evenings and about 40k/s averaged over a day. The busiest single database peaks at 20k changes/s.
• Freshness. Commit to Kafka under 2 s at p99; commit to a fast sink such as cache invalidation under 5 s at p99.
• No loss. A committed change is never lost. Duplicates are allowed (at-least-once delivery), so every sink must be idempotent.
• Order. Per primary key, in commit order. No promise across keys or tables, though transaction boundaries are available to consumers that need them.
• Gentle on the source. A few percent of extra CPU on a primary, at most. The log a stalled reader forces the database to keep must be capped: the pipeline may fall behind, but it must never fill a payments database’s disk.
• Replay. Seven days of full history, and the latest version of every row available indefinitely for bootstrapping.
• Availability. 99.9% for the pipeline — about 43 minutes a month of lag above target. A source failover is absorbed in minutes, without re-copying tables.

Out of scope

• The sinks themselves. We deliver ordered, replayable events and a rule for applying them idempotently; the search indexer and lakehouse loader are their owners’ designs.
• Hand-designed domain events. Events another team builds a product on belong in a transactional outbox. The platform carries them, but their design is not ours.
• Distributed transactions across databases. CDC propagates changes after commit; it does not coordinate writes.
• Stream processing — joins, windows, aggregations. That is Flink and stream processing reading our topics.
• Non-relational sources such as MongoDB or DynamoDB, which have their own change streams.

The set-up

Plan the approach

Build the path for one database first: read its log, publish events, let consumers subscribe. Then add bootstrapping, fan-out and a control plane, ending with a design that works for a handful of databases. Spend most of your time in the deep dives — capture method, snapshots, ordering and duplicates, failover, schemas and catch-up — because that is where CDC systems break in production.

Core entities

• Source. One database cluster: engine, version, primary and standbys, and the tables to capture.
• Source position. Where the reader is in the log. For PostgreSQL, an LSN (log sequence number), a byte position in the write-ahead log. For MySQL, a GTID set or a binlog file and offset. This one value makes resuming, ordering and de-duplication possible.
• Change event. One row change. The key is the primary key; the value holds before, after, op (c create, u update, d delete, r snapshot read) and a source block.
• Table topic. One Kafka topic per captured table, such as cdc.orders.public.orders, partitioned by key.
• Schema version. The Avro schema of a table’s key and value at a point in time, stored in a schema registry and referenced by ID from each event.
• Snapshot job. A request to copy existing rows: table, optional key-range filter, chunk size, last completed key, status.

An update to an order, as the consumer sees it (key { "id": 4711 }):

{
  "op": "u",
  "before": { "id": 4711, "status": "PACKED",  "amount_paise": 129900 },
  "after":  { "id": 4711, "status": "SHIPPED", "amount_paise": 129900 },
  "source": {
    "connector": "postgresql", "db": "orders", "schema": "public", "table": "orders",
    "lsn": 24023128, "txId": 555, "ts_ms": 1793107440123, "snapshot": "false"
  },
  "ts_ms": 1793107440201
}

source.ts_ms is when the change happened in the database; the outer ts_ms is when the connector processed it. Their difference is your capture lag.

API

There are three surfaces. The event contract is the envelope above, one topic per table, with schemas in the registry. Consumers use plain Kafka consumer groups; there is no custom read API. The control API is how teams onboard sources and request snapshots:

POST /v1/sources
Content-Type: application/json

{
  "name": "orders-pg",
  "engine": "postgresql",
  "primary": "orders-pg.internal:5432",
  "tables": ["public.orders", "public.order_items"],
  "bootstrap": "incremental",
  "replay_days": 7
}

{
  "source_id": "src_118",
  "slot": "cdc_orders_pg",
  "publication": "cdc_orders_pg",
  "topics": ["cdc.orders.public.orders", "cdc.orders.public.order_items"],
  "status": "SNAPSHOTTING"
}

POST /v1/sources/src_118/tables     add tables; only the new ones are snapshotted
POST /v1/snapshots                  re-snapshot a table, or a key range with a filter
GET  /v1/sources/src_118/status     lag, retained WAL, last position, slot health
POST /v1/schema-changes:check       DDL in, compatibility verdict out
POST /v1/sources/src_118:pause      stop reading; the database keeps the log meanwhile

Onboarding fails fast if prerequisites are missing: logical decoding not enabled, a table without a primary key, or a connector role without replication rights.

Back-of-envelope estimates

Show the arithmetic (method in back-of-envelope estimation); a few of these numbers decide the design.

Event size. A typical row has 15 columns and is about 300 bytes as JSON. An update carries it twice, plus a source block of about 200 bytes: 300 × 2 + 200 = 800 bytes. Call it 1 KB to cover wide tables. Avro with a registry drops field names and shrinks this, but size for 1 KB.

Ingress. 100k changes/s × 1 KB = 100 MB/s at peak. Averaged over a day it is closer to 40k/s, 40 MB/s.

Retention. Seven days at the average rate: 40 MB/s × 604,800 s ≈ 24 TB, ≈ 73 TB with three replicas. If a whole week ran at peak: 100 MB/s × 604,800 s ≈ 60 TB, ≈ 181 TB replicated. Add the latest version of every row, kept indefinitely for bootstrapping: about 20 billion live rows × 600 bytes ≈ 12 TB, 36 TB with three replicas. Plan around 110 TB, and let tiered storage move segments older than a day to object storage so a bad week does not double the broker count.

Brokers and network. 100 MB/s in becomes 300 MB/s of disk writes with three replicas. With six consumer groups on an average topic, peak reads are 600 MB/s, mostly from page cache. Across 18 brokers that is about 17 MB/s of writes and 33 MB/s of reads each. Bandwidth is not the constraint.

Partitions. Partition count sets consumer parallelism and cannot change later without moving keys to new partitions. The busiest table, orders, peaks at 8k changes/s. If the search indexer applies about 500 documents/s per partition, it needs 8,000 ÷ 500 = 16 partitions; choose 48 for 3× growth. Across the fleet: 1,800 small tables × 3 + 200 busy tables × 16 on average ≈ 8,600 partitions, about 26,000 replicas.

Connector tasks. A database log has one reader, so each database gets one single-threaded task: 200 tasks, packed about 8 per worker with headroom, 30 workers across three zones. The average database peaks at 100k ÷ 200 = 500 changes/s, which is trivial. But suppose your benchmark shows one task sustaining about 10k events/s (measure it — it depends on row width and serialisation). The busiest database, at 20k changes/s, then falls behind by 10k events/s: 36 million events an hour.

What a stalled reader costs the source. That busiest database writes about 1.5 KB of WAL per row change once index entries and occasional full-page images are counted: 20k × 1.5 KB = 30 MB/s. A slot that stops advancing for an hour pins 30 MB/s × 3,600 s ≈ 108 GB of WAL on the primary’s disk.

Bootstrapping. The largest table has 2 billion rows. Throttled to 20k rows/s so the primary barely notices, one reader needs 2B ÷ 20k = 100,000 s ≈ 28 hours; four parallel key ranges cut that to about 7. A job that long must survive restarts.

High-level design

Build the design one requirement at a time. By the end it works for a handful of databases; the deep dives make it work for 200.

Capture every committed change from the database’s log

Every serious database already writes each change to a log before it reports a commit, for crash recovery and for its own replicas. CDC reads that log. On PostgreSQL you set wal_level = logical (a restart), create a publication listing the tables to capture, and create a logical replication slot that uses the built-in pgoutput plugin. The connector’s role needs the REPLICATION attribute and SELECT on those tables. On MySQL you need binlog_format = ROW, binlog_row_image = FULL and, ideally, GTIDs; the connector then connects like a replica with its own server-id and receives the binlog.

1. Commit. The app commits; PostgreSQL writes the change to the WAL before the commit returns.
2. Decode. The slot decodes committed transactions from the WAL, in commit order, for the tables in the publication. A transaction is emitted only once it commits, so rolled-back work never leaks out.
3. Publish. The connector turns each row change into an event keyed by primary key and writes it to that table’s topic.
4. Record. Once Kafka acknowledges the events, the connector stores its position — the LSN — in Kafka Connect’s offsets topic.
5. Confirm. It then confirms that LSN to the slot, telling PostgreSQL it may recycle older WAL.
6. Consume. Consumers read each topic at their own pace.

The slot is the contract. It remembers the last position the connector confirmed, and the database keeps every byte of WAL after it, even across restarts. That makes a crash survivable — and, as a later deep dive shows, makes a dead connector dangerous. Don’t write your own decoder: Debezium provides connectors for both engines and handles the type mapping, snapshots and offsets.

What goes into before depends on a PostgreSQL table setting, REPLICA IDENTITY. By default the old row is logged as its primary key at most, so consumers see the new row but not what it replaced.

Choose FULL for tables whose consumers need old values — the fraud model wants the old and new payout account; a cache keyed by a product’s URL slug must evict the old slug — and leave DEFAULT elsewhere. Watch one more trap under DEFAULT: PostgreSQL stores values over about 8 KB out of line (TOAST), an unchanged TOASTed column is absent from the update, and the connector fills it with a placeholder (__debezium_unavailable_value by default). A sink that writes after blindly overwrites a long product description with that string. On MySQL, the FULL row image the connector requires logs complete before and after rows.

Bootstrap a table, then keep streaming

A new search index needs all 300 million products, not just tomorrow’s price changes, and the log only goes back hours or days. So you copy the existing rows and then switch to the stream. The seam between the two is where CDC systems quietly lose data.

The fix is to tie the copy and the stream to one log position. When a slot is created through the replication protocol, PostgreSQL can export a snapshot that shows exactly the database state after which every change will be in the slot’s stream. The connector reads every row inside a REPEATABLE READ transaction using that snapshot, emits each as an op: r event, commits, and then streams from the slot. MySQL cannot tie a consistent read to a binlog position on its own, so the connector briefly holds a global read lock (table locks on managed services that forbid it) while it records the binlog position and opens a consistent read. Every change after the position arrives after the copy, and nothing in between is lost. This works, but it is one giant transaction, which a deep dive replaces with something gentler.

Fan out to many consumers in per-key order

Each table gets its own topic, and each event’s key is the row’s primary key, so Kafka hashes every version of a row to the same partition. One connector reads the log in commit order and writes each key to one partition, so every consumer sees a row’s versions in the order they were committed. There is no order across partitions or tables, and most sinks don’t need it: a search document or a cache entry depends on one row. (Order per key, and its limits, are covered in message ordering.)

Each consumer is its own consumer group with its own offsets — classic publish-subscribe. The cache invalidator stays at the head, the lakehouse loader reads every few minutes in large batches, and a new fraud feature can start a week back. None of them affects the others, and none of them touches the source database. Topics keep 7 days of every change, so a consumer that is down for a weekend simply resumes.

Onboard databases and evolve schemas without downtime

Hand-editing connector configs does not scale to 200 databases. A control API validates a source (logical decoding on, a primary key on every table, rights granted), creates the publication, slot and topics, and submits the connector to Kafka Connect’s REST API; it also starts snapshots and pauses connectors. A schema registry stores an Avro schema per table topic: the connector registers a new version when a table changes, and consumers fetch schemas by the ID each event carries. A DDL gate in the migration pipeline checks schema changes before they run.

This is the classic control plane versus data plane split. You now have a working CDC platform for a few databases: correct capture, a gap-free bootstrap, ordered fan-out and self-service onboarding. The deep dives are about what breaks at 200 databases and 100k changes a second.

Deep dives

How do we read changes without hurting the source?

There are three ways to learn that a row changed. They differ in what they see and in who pays.

Bad: poll an updated_at column. Every minute, run SELECT * FROM orders WHERE updated_at > :last on every table. It needs no special rights and works on anything that speaks SQL. But deleted rows are simply gone, and three updates inside one interval look like one. Worse, updated_at is stamped when the statement runs, not at commit: a transaction that stamps 10:00:01 but commits at 10:00:09 is invisible to a poll at 10:00:05, and the next poll asks only for later rows — so it is missed forever. And 2,000 tables polled once a minute is about 33 range scans a second, forever.

Good: triggers that write to an audit table. An AFTER INSERT OR UPDATE OR DELETE trigger copies the old and new row into a changes table inside the same transaction, and a relay ships that table to Kafka. It sees every committed change, deletes included. The price lands on the hot path: every captured write becomes two writes inside the customer’s transaction, roughly doubling write latency and WAL for those tables, and triggers must be maintained per table. The relay also inherits a subtle polling bug: sequence IDs are handed out when a row is written, not at commit, so “read IDs above the last one I saw” can skip a slow transaction’s rows.

Great: read the database’s log. The WAL and binlog are written anyway, for crash recovery and replication. Decoding them adds no writes to user transactions, sees every insert, update and delete, emits only committed transactions, and does so in commit order with transaction IDs attached.

Log-based capture is not free either. wal_level = logical needs a restart, scheduled once per database. Each slot’s WAL sender decodes the entire WAL stream, including tables you don’t capture, so several connectors on one server cost real CPU. REPLICA IDENTITY FULL adds WAL. And a slot nobody reads keeps WAL forever, as the failover deep dive shows. On PostgreSQL 16 and later you can decode from a hot standby instead, moving that CPU off the primary at the price of the standby’s lag and extra setup — worth it for the few databases where every percent of primary CPU matters.

How do we snapshot existing rows without gaps or stale overwrites?

The high-level design pinned the copy and the stream to one position. That is correct; the problem is cost at our scale.

Bad: lock the tables and dump them. FLUSH TABLES WITH READ LOCK or LOCK TABLE, dump, note the log position, unlock. Trivially correct, and it blocks every write for the whole dump. For our largest table that is 28 hours of checkout being down. Never on a production primary.

Good: a consistent snapshot at a known position. One REPEATABLE READ transaction reads every row as of position L, without blocking writers thanks to MVCC, and streaming continues from L afterwards. Every row is copied exactly once. But look at what a 28-hour transaction does:

• It holds back VACUUM. PostgreSQL cannot remove dead row versions that an open transaction might still need, anywhere in the database, so the busiest tables bloat for a day (vacuum and bloat).
• It pins the WAL. Streaming waits until the snapshot ends, so the slot keeps every byte since L: 30 MB/s × 28 h ≈ 3 TB on the primary’s disk.
• It can’t resume. If the connector dies in hour 27, the snapshot starts again from zero — Debezium’s documentation says exactly that for a failed initial snapshot.
• Live consumers starve. Every change, even to tables already copied, waits until the whole snapshot finishes.

Great: chunked snapshots between watermarks, interleaved with the live log. Netflix described this in its DBLog paper, and Debezium implements the same idea as incremental snapshots. Instead of one huge read, copy the table in small primary-key chunks while the log keeps streaming, and use two markers written into the log — a low and a high watermark — to work out which copied rows are already out of date.

In your own words, one chunk goes like this:

1. Open the window. Pause log processing in the connector for a moment (the database is not paused) and write a low watermark: update the single row of a tiny watermark table with a fresh UUID. That write enters the log like any other change.
2. Select a chunk. Read the next rows by primary key — WHERE id > :last ORDER BY id LIMIT 1024 — a cheap index range scan, and hold them in a buffer.
3. Close the window. Write a high watermark the same way, then resume log processing.
4. Drop collisions. Keep emitting log events. For every event between the two watermarks, if its key is in the buffer, remove that row from the buffer.
5. Emit the rest. When the high watermark arrives, emit the remaining rows as op: r events, save the chunk’s last key, and start the next chunk.

Why is that safe? The select ran after the low watermark was written and before the high one, so it saw every change committed before the window and perhaps some committed inside it. If a key changed inside the window, the log holds an event for it that is at least as new as the copy, and that event has already gone out — so the copy is either redundant or stale, and dropping it loses nothing. If a key did not change inside the window, the copy equals its state at the high watermark, and any later change follows it in the output. No gap, no stale overwrite, no locks.

What you gain is large. Each chunk is a short query, so nothing holds back VACUUM or the slot; streaming never stops, so live consumers stay fresh during a 28-hour backfill; progress is just “last key copied”, so a crash resumes at the next chunk; and any table, key range or single key can be re-snapshotted on demand. Throttle it by sleeping between chunks. Debezium triggers it with an execute-snapshot signal, written as a row into a signalling table (columns id, type, data) or sent to a Kafka signal topic, and copies 1,024 rows per chunk by default.

What you pay: tables need a primary key with an efficient range scan; the connector needs write access to a signalling or watermark table on the source, although on PostgreSQL 13 and later a read-only variant uses in-flight transaction IDs as watermarks; consumers see r events mixed with live ones and must treat them as upserts; and Debezium’s PostgreSQL connector does not support schema changes while an incremental snapshot runs.

How do we keep per-key order and absorb duplicates?

Order comes from three things lining up. The source log is in commit order; one connector reads it with a single thread; and each key always maps to the same partition, written by Kafka’s idempotent producer, which keeps order within a partition even when it retries. Two edge cases remain. When a primary key itself changes, Debezium emits a delete (and a tombstone) for the old key and a create for the new one; they land in different partitions with no order between them, which is fine for sinks keyed by primary key. And adding partitions to a topic remaps keys, so treat partition count as permanent: to change it, create a new topic and re-snapshot into it.

Duplicates are normal, not exceptional. They come from three places:

• Connector crashes. Kafka Connect commits source offsets periodically — every 60 seconds by default. Events written after the last commit are read again from the slot and written again after a restart.
• Database crashes. A slot’s position is saved only at checkpoints, so after a crash PostgreSQL may resend recent changes.
• Snapshots. r events re-deliver rows that consumers already hold.

The fix belongs in the sink: apply an event only if it is newer than what you hold. Every event carries its source position, so the sink stores the position with each row and compares. For one row, positions always increase in commit order: the row lock forces a second writer to wait until the first commits, so its change is logged later.

INSERT INTO orders_view (id, status, amount_paise, src_lsn, deleted)
VALUES ($1, $2, $3, $4, $5)
ON CONFLICT (id) DO UPDATE
  SET status       = EXCLUDED.status,
      amount_paise = EXCLUDED.amount_paise,
      src_lsn      = EXCLUDED.src_lsn,
      deleted      = EXCLUDED.deleted
  WHERE orders_view.src_lsn < EXCLUDED.src_lsn;

Three details make the rule hold up. Deletes leave a marker: keep a deleted row with its position for a few days, or a late replay of an older update resurrects the row. Snapshot rows need a position too: stamp r events with the high watermark’s position, so later live changes still win. Positions are local: an LSN compares only with LSNs from the same database, and MySQL binlog coordinates restart on a new primary. Where strict idempotence matters on MySQL, or across shards, carry a row_version column the application increments and compare that. Search engines have this built in: write each document with its position as an external version and older versions are rejected. This is the idempotent consumer pattern applied to rows.

Can we get exactly-once instead? Into Kafka, yes. Kafka Connect 3.3 and later can write a source connector’s records and offsets in one Kafka transaction and fence off zombie tasks (KIP-618): set exactly.once.source.support=enabled on every worker and exactly.once.support=required on the connector, and Debezium’s PostgreSQL, MySQL, SQL Server, Oracle, MongoDB and MariaDB connectors take part. You pay with consumers reading isolation.level=read_committed, a little latency because records appear only when each transaction commits, and open Kafka transaction bugs that the Debezium documentation itself flags. And it covers only the hop into Kafka; a sink writing to Elasticsearch still needs the idempotent rule. Use it for Kafka-to-Kafka pipelines that count things, such as a Flink job computing GMV per minute.

How do we survive failovers, crashes and stalled slots?

A connector or worker crashes. Kafka Connect runs in distributed mode across 30 workers. When a worker dies, its tasks restart on other workers and resume from the last committed offset; the replayed events are absorbed by sinks as above. The slot does not care which worker connects, only that one does at a time.

The source fails over. Here PostgreSQL versions matter. Up to PostgreSQL 15, a logical slot exists only on the primary. After a promotion the new primary has the WAL but no slot, and a new slot starts from “now” — every change between the last confirmed position and the promotion is missing from the stream, and Debezium’s advice is a fresh snapshot. PostgreSQL 16 allows slots on a standby, but syncing them is manual. PostgreSQL 17 added failover slots: create the slot with the failover option (Debezium’s slot.failover=true), set sync_replication_slots = on, hot_standby_feedback = on and a primary_slot_name on the standby, and a slotsync worker keeps the standby’s copy current. Only slots whose synced flag was true before the failover are usable afterwards, so alert on it.

The second step is the subtle one. With asynchronous replication, a failover can lose the last few transactions the old primary committed but never shipped. If the connector had already published them, Kafka would hold changes the database no longer has — phantom orders downstream. Listing the standby’s physical slot in synchronized_standby_slots on the primary makes logical decoding wait until the standby has the WAL, so the stream can never get ahead of the database you would fail over to. You pay a little latency on every change. Some managed services replicate the whole disk to the standby, and the slot survives a failover with no extra setup.

MySQL is simpler because of GTIDs. Every transaction gets a global ID (source server UUID plus a sequence number), and the connector stores the set it has processed. After a failover it connects to the new primary and asks for everything not in that set; this works as long as the new primary has the same GTID history. The binlog must still hold what the connector needs: binlog_expire_logs_seconds defaults to 30 days, which comfortably covers a connector outage. If the needed binlog has been purged, the only way back is a snapshot.

A slot stalls. This is the failure that hurts the source. A slot keeps every byte of WAL after its confirmed position, and by default there is no limit.

Unbounded, a stalled slot fills the primary’s disk and takes the database down. It also holds back cleanup of system catalog rows, and the PostgreSQL documentation warns that in extreme cases the database shuts down to prevent transaction ID wraparound. So the platform enforces a policy per database:

• Alert early. Page when a slot retains 50 GB — about 28 minutes into a stall on the busiest database.
• Cap it. Set max_slot_wal_keep_size (PostgreSQL 13 and later; unlimited by default) to 200 GB, about two hours on the busiest database. Past it, PostgreSQL invalidates the slot rather than filling the disk. You choose a re-snapshot over an outage of the payments database, and the control API opens that re-snapshot job automatically.
• Clean up orphans. PostgreSQL 18’s idle_replication_slot_timeout invalidates slots unused for a set time — leftovers of connectors deleted without dropping their slot.
• Send heartbeats. On a quiet database sharing a busy server, the connector receives nothing for its tables, never confirms a newer position, and the shared WAL piles up. Debezium’s heartbeat.interval.ms and heartbeat.action.query write a row to a small heartbeat table every few seconds, giving it something to confirm — and giving you true end-to-end lag even with no traffic.

How do we change schemas without breaking consumers?

First, how the connector even learns that a table changed. PostgreSQL’s logical decoding does not emit DDL. Instead, pgoutput sends a Relation message describing a table’s columns before the first change to it, and again whenever its definition has changed, and the connector refreshes its picture of the table. MySQL writes DDL into the binlog as statements, so the connector parses every ALTER TABLE and stores it, with its binlog position, in a schema history topic — one partition, for the connector only. On restart it replays that topic up to its resume position to rebuild each table’s structure as it was at that point in the log. Lose the topic and old binlog events can no longer be interpreted: protect it like the offsets topic.

The connector turns each table’s structure into an Avro schema and registers it in the schema registry; every event carries the schema’s ID, so consumers can always decode old and new events. The registry enforces a compatibility mode per topic. The default, BACKWARD, means a consumer using the new schema can read data written with the old one: you may add optional fields and remove fields (Confluent’s compatibility rules list them per format). More on the general idea in serialization and schema evolution.

Here is the twist: in CDC, the producer upgrade is the DDL. Whoever runs ALTER TABLE on the orders database changes the contract for the search, fraud and lakehouse teams at that instant. So the check has to happen before the DDL runs, not when the connector registers the result.

Safe changes pass straight through: add a nullable column or one with a default, add a table, widen a varchar. Breaking changes are blocked: renaming a column (to every consumer it is a drop plus an add, so half the code suddenly reads nulls), narrowing a type (bigint to int, or changing a decimal’s scale), dropping a column someone reads, or changing the primary key (which changes the topic key and so the partitioning). Blocked changes are rewritten as expand and contract steps — online schema and data migrations in detail. To rename phone to mobile: add mobile, have the app write both, backfill in small batches, move consumers to mobile, stop writing phone, and drop it once no consumer reads it. MySQL online-schema-change tools such as gh-ost copy into helper tables and swap them in; capture those helper tables too, or filter them out deliberately.

Should other teams read raw table topics at all? Raw table CDC publishes your internal schema as a public contract: every column rename becomes a cross-team negotiation, and consumers see half-finished states, such as an order whose items haven’t arrived. A good rule: raw table CDC for derived copies of your own data (search, caches, the lakehouse), an outbox for contracts between teams. With the transactional outbox, the service writes a deliberate event — OrderPlaced, items included — into an outbox table in the same transaction as the business change. The same pipeline captures it, and Debezium’s outbox event router sends each row to a topic named after its aggregatetype, such as outbox.event.order, with the event ID as a header for de-duplication. Only the event schema is a contract; the tables can change freely.

How do slow and new consumers catch up?

A consumer that was down for a long weekend, or a brand-new one that needs every row, must catch up without hurting the source.

Bad: keep every change forever. Simple, and storage grows by 40 MB/s × 86,400 s ≈ 3.5 TB a day before replication, forever. A new consumer must replay years of changes to rebuild today’s state.

Good: re-snapshot on demand. Trigger an incremental snapshot of the table, or of a key range. It works for any consumer at any time. But each new consumer costs the source a full scan — 28 hours for the largest table — and the r events land in the shared topic, where every existing consumer has to process them too, unless you snapshot into a separate topic for that one consumer.

Great: one topic with two zones — a full 7-day head and a compacted tail. Set cleanup.policy=compact with min.compaction.lag.ms of 7 days. Kafka guarantees that a record is not compacted until it is at least that old, so the last 7 days hold every change for replay. Behind that, compaction keeps the latest record for each key, so the tail is a ready-made snapshot of the table. Deletes are tombstones — a key with a null value, which Debezium emits after each delete by default — and compaction removes them after delete.retention.ms.

A new consumer reads from offset 0: first the latest version of every row, then every recent change, and it is caught up without a single query on the source. A consumer 9 days behind lands in the compacted tail; it skips intermediate versions but still ends in the right state. That is fine for state consumers like search and caches; an audit consumer that needs every version reads the lakehouse instead. One setting deserves care: raise delete.retention.ms from its default of 24 hours to 7 days, because a bootstrap that takes longer than that to reach a tombstone can miss the delete and keep a row that no longer exists. For the biggest tables, the lakehouse copy is a cheaper bootstrap still: load its files, then consume from the Kafka offsets it recorded with its last commit.

This is not a new idea. LinkedIn’s Databus paired relays that held the recent change stream with a bootstrap service that kept a moving snapshot by applying that stream. A consumer that fell off the relay got a consolidated set of changes; a new one got a consistent snapshot and then continued from the relay — all without touching the primary database. A compacted topic with a lag floor builds the same two tiers into Kafka.

The final design

Put the pieces together. Changes flow left to right; everything below the line configures and guards that flow without sitting on it.

Follow one price change through the system:

1. Commit. The catalogue service commits the UPDATE. The change lands in the WAL and streams to the standby, which holds a synced copy of the CDC slot.
2. Read. The database’s connector — one single-threaded task on a 30-worker Kafka Connect cluster — receives the committed transaction through its slot, or as a binlog replica on MySQL.
3. Publish. It builds the Avro envelope, registers a schema version if the table changed, and writes the event with an idempotent producer to the table’s topic, keyed by primary key.
4. Confirm. Once Kafka has the event and the offset is committed, it confirms the position so the database can recycle WAL.
5. Consume. Each consumer reads at its own pace and applies the event only if it is newer than what it holds. The topic keeps every change for 7 days and the latest row per key after that.
6. Snapshot. The control API onboards sources and starts watermark snapshots with a signal; chunks interleave with the live stream.
7. Guard. The monitor turns heartbeats into true lag and tracks WAL kept per slot: alert at 50 GB, cap at 200 GB, re-snapshot on invalidation. Failover slots (PostgreSQL 17) and GTIDs (MySQL) let the connector resume on a promoted standby.
8. Change. Every migration passes the DDL gate, which checks the new schema with the registry before the DDL runs.

Check it against the requirements: every committed change, with old values where configured; bootstraps that never lock the source or pause the stream; per-key order with duplicates made harmless; 7 days of replay plus the latest row per key; 100 MB/s of peak ingress on 18 brokers sized mainly for storage; commit-to-Kafka lag measured by heartbeats against the 2 s target, with the one database too hot for a single reader called out; and a source whose cost — decoding CPU and at most 200 GB of WAL — is bounded and watched.

What interviewers look for by level

Follow-up questions

What if someone runs a 50-million-row UPDATE in one transaction?

Logical decoding emits a transaction only when it commits, spilling decoded changes to disk beyond logical_decoding_work_mem (64 MB by default). Then 50M events arrive at once: about 50 GB into Kafka, and at 10k events/s per task, 50M ÷ 10k = 5,000 s — roughly 83 minutes during which every other table in that database lags. Prevent it: the migration gate rejects unbatched bulk updates and runs backfills in transactions of about 10k rows, throttled and off-peak, with consumers warned.

What if one database does 20,000 changes a second?

One task can’t keep up, so first make it faster: Avro instead of JSON, larger producer batches, compression, no transforms on the hot path, and only the tables people need. Next, split the tables across two connectors with separate slots. Each slot decodes the whole WAL, so decoding CPU on the primary doubles, and the two halves lose commit order relative to each other — keep tables written in the same transactions together. The lasting fix is sharding the database, which it probably needs anyway.

What if a customer asks us to erase their personal data?

Laws such as India’s DPDP Act and the EU’s GDPR give people a right to erasure, and CDC copies personal data everywhere. Deleting the source row emits a delete and a tombstone, but older values sit in the 7-day head until compaction, and in the lakehouse’s history. Encrypt personal columns in the connector with a per-customer key, and erase by deleting the key — crypto-shredding — so every copy, including backups, becomes unreadable at once (encryption and key management). The lakehouse also runs row-level deletes and expires old snapshots.

What if a consumer needs a consistent view across tables?

An order and its items commit together but land in different topics, so a consumer can see the order first. Turn on transaction metadata: Debezium then writes BEGIN and END events to a transaction topic, the END carrying event counts per table, and stamps each change with its transaction ID and position within it. A stream processor buffers by transaction ID until the counts match, then emits the whole transaction. You pay memory and the latency of the largest transaction. Often the better answer is an outbox event that already contains the order and its items.

What if the orders database is split into 64 shards?

That is 64 connectors and slots, with a routing transform merging every shard’s orders table into one topic keyed by a globally unique order ID. Per-key order holds while each key lives on one shard. During resharding, though, a moved row is deleted on one shard and inserted on another by two connectors with no order between them, and their LSNs are not comparable. So sinks compare a row_version that moves with the row, and the resharding tool marks rows it moves so the router drops their deletes.

How to say it in the interview

“I’ll design a company-wide CDC platform: 200 PostgreSQL and MySQL databases, 2,000 tables, 100 thousand row changes a second at peak, with no committed change ever lost and the source barely noticing.

I’d read each database’s own log, not poll tables or add triggers. On PostgreSQL that’s a logical replication slot with pgoutput; on MySQL, the row-format binlog with GTIDs. One Debezium connector per database runs on a Kafka Connect cluster and turns each committed row change into an Avro event with before and after images and its source position, written to a per-table topic keyed by primary key, so every row’s changes stay in commit order. The connector confirms a position to the database only after Kafka has the events.

To bootstrap a table, I use chunked snapshots between low and high watermarks written into the log: any chunk row whose key changes inside the window is dropped because the log already has a newer version. It’s lock-free, resumable and runs while streaming continues.

Delivery is at least once. Crashes and snapshots replay events, so every sink applies an event only if its position, or a row version, is newer than what it holds, and keeps delete markers. Exactly-once into Kafka exists, but the idempotent sink is what makes it safe end to end.

For failures: PostgreSQL 17 failover slots, with decoding held back until the standby has the WAL so nothing published can vanish; GTIDs on MySQL. Retained WAL is alerted at 50 gigabytes and capped at 200, trading a re-snapshot for a full disk, and heartbeats keep quiet databases from pinning WAL.

Schema changes go through a DDL gate that checks the registry before the ALTER runs, with expand-and-contract for renames, and outbox events for cross-team contracts. Topics are compacted with a 7-day minimum lag, so they hold a week of replay plus the latest row per key, and new consumers bootstrap from Kafka, not from the database.”

Concepts used in this design

• Change data capture — the concept this platform implements at scale.
• Write-ahead logging — the log CDC reads instead of the tables.
• Kafka — partitions, consumer groups, compaction and Connect.
• Message ordering — per-key order through partitioning.
• Delivery guarantees — at-least-once capture and where exactly-once stops.
• Idempotent consumers — applying only newer versions.
• MVCC and vacuum and bloat — why long snapshots hurt PostgreSQL.
• Data replication and failover — standbys, promotion and what a failover can lose.
• Serialization and schema evolution — Avro, registries and compatibility.
• Online schema and data migrations — expand and contract for breaking changes.
• Transactional outbox — deliberate events for contracts between teams.
• Control plane vs data plane — onboarding and gating off the change path.
• Database sharding — splitting a database that one log reader can’t keep up with.

References and further reading

1. DBLog: A Watermark Based Change-Data-Capture Framework (Netflix, arXiv 2010.12597)arxiv.orgThe low and high watermark algorithm: select a primary-key chunk between two watermark writes, drop chunk rows whose keys change inside the window, emit the rest at the high watermark, resume from the last completed key.
2. Debezium connector for PostgreSQLdebezium.ioReplica identity and before images, TOAST placeholders, incremental snapshots with 1,024-row default chunks, heartbeats to release WAL, one task per connector, failover slots on PostgreSQL 17, duplicates after a crash.
3. Debezium connector for MySQLdebezium.ioRequired binlog settings (ROW format, FULL row image, 30-day default expiry), GTIDs, snapshot locking, and the single-partition schema history topic that rebuilds table structure from DDL.
4. Debezium: Exactly once deliverydebezium.ioAt-least-once by default; which connectors join Kafka Connect exactly-once (Connect 3.3 or later), the worker and connector settings, and open Kafka transaction caveats.
5. PostgreSQL documentation: Logical decoding conceptspostgresql.orgSlots survive crashes but may resend changes, pin WAL and catalog rows even when idle, synchronise to standbys as failover slots, and export a snapshot for a gap-free initial copy.
6. PostgreSQL documentation: Replication settingspostgresql.orgmax_slot_wal_keep_size (unlimited by default), idle_replication_slot_timeout, sync_replication_slots and synchronized_standby_slots.
7. Apache Kafka design: Log compactionkafka.apache.orgCompaction keeps at least the latest value per key, tombstones, delete.retention.ms (24 hours by default) and min.compaction.lag.ms as a floor on uncompacted history.
8. Apache Kafka: Kafka Connect configskafka.apache.orgSource offsets stored in a topic and flushed every 60 seconds by default; exactly.once.source.support and its preparing-then-enabled rollout.
9. Confluent Schema Registry: Schema evolution and compatibilitydocs.confluent.ioBACKWARD as the default mode, what backward, forward, full and transitive modes allow, and which field changes each accepts.
10. Open sourcing Databus: LinkedIn’s low latency change data capture systemengineering.linkedin.comRelays plus a bootstrap service that keeps a moving snapshot, so new or far-behind consumers catch up without loading the source database.