Lesson 4 · CDC foundations & the source

The change event & topics

What Debezium actually puts on the wire — the envelope, the Avro encoding, the topic per table, and how a delete looks.

Your win: read a Debezium change event, explain why it's serialized as Avro through a Schema Registry, know how topics are named, and recognize a tombstone.

The change event: the envelope

For every row change, Debezium emits one message with a consistent shape — the envelope:¹

the Debezium envelope (conceptual)

{
  "before": { … row as it was …  } | null,   // null for inserts
  "after":  { … row as it now is … } | null,   // null for deletes
  "op":     "c" | "u" | "d" | "r",         // create / update / delete / read(snapshot)
  "source": { db, schema, table, lsn, txId, ts_ms, … }   // metadata: where & when
}

So an insert has before: null, an update has both, a delete has after: null, and a snapshot row carries op: "r". The destination usually only wants the new state — so a sink transform (unwrap, Lesson 6) flattens this to just the after row. But the full envelope is what travels on the topic.

One topic per table — and the naming

Debezium writes a separate Kafka topic per captured table. By default that's <prefix>.<schema>.<table> (e.g. bob.public.users) — but this repo rewrites it with a routing transform on the source so every topic gets an env.org prefix:

local/…/kafka-connect-source-connector/bob.json:42-45

transforms                 = "route"
transforms.route.type      = "org.apache.kafka.connect.transforms.RegexRouter"
transforms.route.regex     = "([^.]+).([^.]+).([^.]+)"       // bob . public . users
transforms.route.replacement = "local.manabie.$1.$2.$3"       // → local.manabie.bob.public.users

So the topic becomes local.manabie.bob.public.users — {env}.{org}.{db}.{schema}.{table}. That five-part name is why the sink's route regex later extracts $5 to recover the bare table name (Lesson 6). Naming is load-bearing here.

Avro + the Schema Registry

The events aren't JSON on the wire — they're Avro, via a Confluent Schema Registry:

source.go:160-165

value.converter                    = "io.confluent.connect.avro.AvroConverter"
value.converter.schema.registry.url = "http://cp-schema-registry:8081"

Why Avro through a registry Avro is compact (binary, no field names repeated per message) and schema'd. But putting the full schema in every message would be huge — so the Schema Registry stores schemas centrally, and each message carries just a small schema ID. The sink looks the ID up to decode. It also enables schema evolution: add a column, register a new compatible version, and consumers keep working. (Note: this repo sets registry compatibility to none — it doesn't enforce compatibility checks, trading safety for flexibility.)

How a delete looks: the tombstone

A delete produces two messages: the delete event (op: "d", after: null), then a tombstone — a message with a null value (tombstones.on.delete = true). The tombstone tells a log-compacted topic it may garbage-collect that key.²

But deletes don't reach the destination here Foreshadowing Lesson 6: the sink connectors are configured to drop deletes and tombstones (delete.enabled = false). Instead of deleting the destination row, the platform relies on soft deletes — a deleted_at column that's just another column update, which does propagate. So a "delete" flows through as an update setting deleted_at, and the hard-delete event is discarded. Consistent with the rest of the codebase's never-hard-delete rule.

(Topics are also created with a compacting, lz4-compressed, 10-partition default — bob.json:37-41 — but that's tuning; the envelope, naming, Avro, and tombstones are the concepts that matter.)

The Debezium change event & Avro serialization

The envelope structure in detail, and how the Avro converter + Schema Registry encode it.

→ debezium.io — change events
→ Confluent — Avro serdes · Schema Registry

Check yourself (from memory)

Q1. In the Debezium envelope, what does an insert look like?

Insert = no prior state (before: null), the new row in after, op c. Delete is the mirror; update has both.

Q2. Why are change events serialized as Avro through a Schema Registry?

Schemas live centrally in the registry; each message references a small schema ID. Compact, and it enables schema evolution.

Q3. What is a tombstone, and what happens to it here?

A null-value message enabling log compaction to drop the key. This repo drops deletes/tombstones and soft-deletes via deleted_at instead.

Recall: the change event, topics, Avro, and tombstones.

envelope + naming + Avro + delete, then reveal

Envelope: per change — before/after (row state; null for insert/delete) + op (c/u/d/r=snapshot) + source (db/schema/table/lsn/ts). Sink unwrap later flattens to after. Topics: one per table; source RegexRouter renames bob.public.users → local.manabie.bob.public.users ({env}.{org}.{db}.{schema}.{table}). Avro + Schema Registry (cp-schema-registry:8081): compact binary; message carries a schema ID, not the schema; enables evolution (compatibility set to none here). Delete = an op:d event + a tombstone (null value, for log compaction) — but sinks drop deletes; platform soft-deletes via deleted_at (a normal update that does propagate).

🏁 Part 1 complete — CDC foundations & the source You can now explain the capture side end to end: what CDC is and the three stacks (L1); how Debezium reads the WAL via pgoutput, slots, and publications (L2); snapshots and the add-a-table-without-recreation trick (L3); and the change event, Avro topics, and tombstones (L4). The row change is now an event on a topic. Part 2 carries it to its destination.

Part 2 opens the Kafka Connect framework and the sink side — the runtime, the JDBC upsert sinks, the connector-generation system, and hephaestus. Tell me "build Part 2" when you're ready, or ask me anything about the source first. Re-take these four quizzes tomorrow cold — spacing is what makes it stick.

1. Debezium — change events. In-repo: local/configs/services/kafka-connect-source-connector/bob.json:42-45.

2. Confluent — Avro. In-repo: internal/platform/datapipeline/source.go:160-171.