CDC with Kafka and Debezium

Data Engineering & Modern Data Platforms · Lesson 46 · Real-Time Data Engineering and Kafka

Lesson Overview

The cohort counter still has the requirement from Change Data Capture Fundamentals. Instructor operations need to know who is enrolled now, including a hard unenrollment, within about fifteen minutes. A poll of updated_at missed deletes and missed updates that did not touch the watermark column. Lesson 16 stopped at the mechanism: PostgreSQL logical decoding, a replication slot, replica identity, a snapshot coordinated with the slot, and a fold that applies changes in log-sequence-number order. It did not deploy a connector. This lesson deploys one.

Debezium's PostgreSQL connector is a Kafka Connect source. It uses the logical decoding plugin you choose, pgoutput in this lesson, and it writes a change event per insert, update, delete, and snapshot read to a topic. The envelope is Debezium's documented JSON or Avro structure: before, after, source, and op. The source.lsn field is the same order key lesson 16 folded. The cohort job does not grow a second ordering rule because a connector arrived.

Kafka Connect and Data Integration already showed why the JDBC source is the wrong tool here. Incrementing mode never emits a delete. Enrollment's product is the delete. The JDBC connector stays on purchase_line. Debezium stays on enrollment. Both are Connect plugins. Only one of them reads the write-ahead log.

Two cursors appear, and confusing them fills the primary's disk or drops changes. PostgreSQL remembers a slot restart LSN and retains WAL until the slot moves. Connect remembers a source offset, the LSN the connector has committed into connect-offsets. The connector is healthy only when both move. A stopped task leaves the slot where it was. The primary keeps the WAL. Lesson 16 named that failure. This lesson gives you the query and the connector settings that make it visible.

/learn/kafka is the log the connector produces into. /learn/postgresql is the engine it reads. Nothing in this lesson is a private TechLearningPro connector. Class names and envelope fields follow the Debezium PostgreSQL connector documentation. Property names below match Debezium 2.x (topic.prefix rather than the older database.server.name). Debezium 3.x kept that prefix. Confirm snapshot mode names against the version you install before you paste a config into production.

Learning Objectives

By the end of this lesson you will be able to:

1. Map Debezium's snapshot and streaming phases onto lesson 16's snapshot-plus-slot handshake.
2. Configure the PostgreSQL connector with pgoutput, a publication limited to enrollment, and a replication slot you can name.
3. Read op values r, c, u, and d, and use source.lsn as the order key the fold already expects.
4. Choose REPLICA IDENTITY DEFAULT for this table and say when FULL would be required instead.
5. Explain the tombstone after a delete, and why the cohort fold must apply op=d and not only the tombstone.
6. Monitor retained WAL on the slot and treat a stopped task as a disk incident in progress.
7. Keep Connect offsets and the slot as a pair: do not delete one and keep the other without a plan.
8. Fold the envelope with the lesson 16 rules: per-key LSN order, idempotent replay, no resurrection.
9. Keep CDC deletes off historical completion facts.
10. Say what the connector does not make exactly-once in the cohort table, and what the LSN check still has to do.

Prerequisites

• Change Data Capture Fundamentals: query versus trigger versus log, replica identity, snapshot coordination, LSN ordering, and the resurrection bug. This lesson assumes that fold exists.
• Lesson 42: distributed Connect, the offset topic, errors.tolerance, and why JDBC incrementing mode cannot see deletes.
• Relational Database Internals: the write-ahead log is how PostgreSQL recovers. Logical decoding reads it. You do not need a new theory of pages.
• Kafka Consumer Groups and Offset Management: the cohort consumer has its own offsets. Those are a third cursor, after the slot and the Connect offset.
• /learn/postgresql for roles, publications, and ALTER TABLE ... REPLICA IDENTITY. /learn/kafka for the topic that will hold the envelope.

Detailed Core Concepts

What the connector actually reads

PostgreSQL with wal_level=logical can decode committed changes through an output plugin. pgoutput is the plugin shipped with the server. It is also what PostgreSQL's own logical replication uses. Debezium can use it without a custom .so on the host. Older tutorials install decoderbufs. You do not need that plugin for this design, and adding extensions to the primary is a change you make only when pgoutput cannot do something you measured.

The connector creates or attaches to a replication slot. The slot is the server-side memory of how far decoding has confirmed. WAL before the slot's restart LSN can be recycled. WAL after it cannot. If the Connect task is down, confirmation stops, and the primary's disk fills until PostgreSQL starts refusing writes. That is the operational price of seeing deletes. A nightly snapshot does not have this price. The cohort product's fifteen-minute bound is why the price is worth paying for enrollment and not for append-only progress events. Lesson 16 already declined a slot on the progress table. Do not add one because the connector makes it easy.

A publication lists the tables whose changes pgoutput will stream. CREATE PUBLICATION enrollment_pub FOR TABLE public.enrollment is the boundary. publication.autocreate.mode=disabled stops the connector from creating a publication for every table because a config default said so. The DBA owns the publication. The connector user uses it.

Snapshot, then the stream

A new slot does not contain the rows already in enrollment. The cohort table needs those rows. Debezium's snapshot.mode=initial is the handshake from lesson 16, implemented inside the connector:

1. The connector starts a snapshot consistent with a point in the log.
2. It emits one event per existing row with op=r (read). source.snapshot is set on those events.
3. It then streams commits that were not part of the snapshot, with op=c, op=u, and op=d.
4. It records the LSN in the Connect offset so a restart continues rather than snapshotting again.

snapshot.mode=never is for the case where you already loaded a snapshot by some other coordinated path and you accept responsibility for the LSN handshake. initial_only snapshots and stops, which is a bulk load wearing a CDC connector. when_needed can snapshot again if the offset is gone and the slot cannot resume. Read that mode as "I might snapshot in production when I did not plan to." Prefer initial for the first deploy and a deliberate incremental snapshot later.

Debezium's incremental snapshot is signaled by a row in a signal table, not by dropping the slot. The table the connector documents is a primary key id, a type, and a data payload. You insert a row of type execute-snapshot naming public.enrollment. The connector reads it because signal.data.collection points at that table. Use this when you need to re-copy enrollment without abandoning the slot and without a second full initial snapshot racing the stream. The fold must treat op=r as idempotent on the key and the LSN, or a second snapshot doubles nothing only because you ignore LSNs you have applied. Lesson 16's rule still holds: same LSN and same payload is a no-op; same LSN and a different payload is a breach.

A long open transaction is invisible until it commits, and then it arrives as a burst. The fifteen-minute cohort bound is not a bound on transactions that were open for an hour. Say that in the contract, as lesson 16 did. The connector does not change commit visibility.

The envelope and the order key

A streaming update looks like this in structure. Field names are Debezium's. Values are illustrative.

jsonCopy
{
  "before": {"learner_id": "l-1", "course_id": "course-a", "status": "enrolled"},
  "after": {"learner_id": "l-1", "course_id": "course-a", "status": "completed"},
  "source": {
    "connector": "postgresql",
    "name": "enrollment",
    "db": "techlearningpro",
    "schema": "public",
    "table": "enrollment",
    "lsn": 24012345,
    "txId": 9012,
    "ts_ms": 1740000000000,
    "snapshot": "false"
  },
  "op": "u",
  "ts_ms": 1740000000100
}

op is c for insert, u for update, d for delete, r for a snapshot read, and t for truncate if you ever allow truncate on this table. Do not truncate enrollment as an operational habit. A truncate event is a delete of every key, and a consumer that ignores t will keep the cohort full of people the table no longer has.

source.lsn is the log sequence number. Apply per (learner_id, course_id) in increasing LSN order. A delete at a higher LSN must not be overwritten by an update at a lower LSN that a parallel consumer handled later. Partition the cohort apply by the enrollment key, or apply serially per key. That is the resurrection bug from lesson 16, and Kafka partitioning does not save you if the topic key is wrong. Set the message key to the primary key, which Debezium does by default: a struct of learner_id and course_id. Give the topic enough partitions for throughput, and accept that one key stays on one partition so a single consumer of that partition sees the key in order. Ordering across keys is not required.

source.ts_ms is the database time of the commit. ts_ms at the root is when the connector processed the event. Freshness dashboards should say which one they use. The cohort's "fifteen minutes" should be measured from commit time, not from connector time, or a stalled connector looks fresh until you notice the slot.

Replica identity and the before image

REPLICA IDENTITY DEFAULT uses the primary key. For enrollment the key is (learner_id, course_id). A delete event can carry that key. With pgoutput and DEFAULT, the before image on update and delete contains the key columns. It does not contain the old status. The cohort fold needs the key on a delete and the new status on insert and update. It does not need the old status. DEFAULT matches lesson 16's choice.

REPLICA IDENTITY FULL puts the entire old row in the log. Use it when a consumer must compute a delta from the before image. It increases WAL volume on every update. Do not set it because a tutorial did. NOTHING records no old identity. Deletes become hard to apply. If you find NOTHING on enrollment, the cohort will "miss" unenrollments in a way that looks like a connector bug and is a table setting.

Changing replica identity does not rewrite WAL you already discarded. It affects later changes.

Tombstones

With tombstones.on.delete=true, which is the connector default, a delete produces two records: the op=d event, and then a record with the same key and a null value. The null value is a Kafka tombstone so a compacted topic can eventually drop the key. The cohort fold must apply op=d as "remove this key from current state." If it only listens for tombstones, it has the key and not the LSN that was inside the value, unless you copied the LSN somewhere else. If it ignores tombstones and applies op=d, compaction can still clean the topic later and the fold has the LSN. Apply op=d. Treat a tombstone with no preceding delete as a delete of that key only if you might have lost the op=d record, and record that you did not have an LSN. Do not let a tombstone resurrect a row. A null value is not an enroll.

Compaction is optional on the enrollment change topic. If you compact, you must understand that intermediate updates can disappear and the latest key remains, which is current state, not history. The cohort wants current state. A replay of history wants the uncompacted log, with retention long enough to rebuild. Those are two topics or two policies. Do not compact the only copy and also promise a rebuild of yesterday's transitions.

Three cursors

If you drop the slot and keep the Connect offset, the connector may fail to resume or may start a new slot at the end and miss the gap. If you delete the Connect offset and keep the slot, snapshot.mode decides whether you snapshot again or try to continue. Either choice is a plan you write down, not a cleanup script. If the cohort consumer rewinds and the fold is idempotent on LSN, you replay safely. If the fold is not idempotent, a rewind double-applies a status. Lesson 16's fold is the safety. The connector does not replace it.

Connect source exactly-once (the worker setting exactly.once.source.support, Kafka 3.3 and later) can make the produce of a change and the offset commit transactional when the connector declares support. Check the Debezium release notes for the PostgreSQL connector version you run before you promise that mode. Even when it is on, the cohort table is a second write outside that transaction. A crash after Kafka has the event and before the cohort commit will redeliver. The LSN check makes that redelivery a no-op. I would rather have the LSN check and a clear at-least-once story than a configuration I did not verify.

What stays off this connector

Progress events remain a produced topic, learning.progress, landed by the sink from lesson 42 and windowed in lesson 45. A slot on that table would retain WAL for a high-volume append-only feed the estate already has. Purchase lines stay on the JDBC incrementing source because they are insert-only and the database is not required to emit deletes it forbids. Enrollment is the table whose deletes are the product.

Historical completion facts do not subscribe to op=d. A learner who completed a lesson on Wednesday and was deleted from enrollment on Friday remains complete for Wednesday. The cohort counter on Friday does not show them. One "apply CDC to the warehouse" job will either destroy history or fail to drop the cohort. Two consumers, two meanings, one topic.

Architecture and Design Diagrams

mermaidCopy
flowchart LR
    subgraph pg [PostgreSQL primary]
        table[enrollment]
        wal[WAL wal_level logical]
        slot[Slot enrollment_debezium]
        pub[Publication enrollment_pub]
        table --> wal --> slot
        pub --> slot
    end

    subgraph connect [Connect worker]
        deb[Debezium Postgres connector]
        off[(connect-offsets LSN)]
    end

    subgraph kafka [Kafka]
        topic[enrollment.public.enrollment]
    end

    subgraph consumers [Separate consumers]
        cohort[Cohort fold current state]
        hist[Daily snapshot job not this slot]
        mart[Completion facts]
    end

    slot --> deb
    deb --> off
    deb --> topic
    topic -->|apply op d to current state| cohort
    table -->|scheduled snapshot| hist
    hist --> mart
    cohort -.->|must not delete| mart

The daily snapshot path does not read the slot. It is the same split as lesson 16. The connector replaces the hand-rolled decoder, not the product split.

mermaidCopy
sequenceDiagram
    participant PG as PostgreSQL
    participant C as Debezium task
    participant K as Kafka topic
    participant F as Cohort fold

    C->>PG: Start snapshot consistent with an LSN
    PG-->>C: Current enrollment rows
    C->>K: op r for each row, with source.lsn
    C->>K: Commit Connect offset at snapshot LSN
    PG-->>C: Later commit update then delete
    C->>K: op u then op d then tombstone
    K->>F: Deliver in partition order
    Note over F: Apply increasing LSN per learner_id course_id
    F->>F: op d removes current state only

A tombstone with a null value follows op=d. The fold's decision is on op and lsn. The tombstone exists so compaction can drop the key.

Technical Explanation

Server settings and roles

wal_level=logical requires a restart of PostgreSQL. max_replication_slots and max_wal_senders need a free slot and a free sender for this connector, plus whatever replicas you already run. Take the slots from a counted budget. A second "temporary" connector that creates another slot is how two abandoned slots fill the disk.

sqlCopy
CREATE ROLE debezium_repl WITH REPLICATION LOGIN PASSWORD '...';
GRANT SELECT ON TABLE public.enrollment TO debezium_repl;

CREATE PUBLICATION enrollment_pub FOR TABLE public.enrollment;

ALTER TABLE public.enrollment REPLICA IDENTITY DEFAULT;

The primary key on (learner_id, course_id) has to exist or DEFAULT has nothing to log. Lesson 16 was explicit. The connector will not invent a key. SELECT and REPLICATION are the privileges this role needs for pgoutput when the publication already exists. Do not grant superuser because a guide from 2018 did. On the PostgreSQL version you run, confirm any extra grant the Debezium page lists for that version, especially around publication ownership. The publication is created by a migration role, not by the connector, because autocreate is disabled.

The signal table, if you want incremental snapshots:

sqlCopy
CREATE TABLE public.debezium_signal (
    id varchar(255) PRIMARY KEY,
    type varchar(32) NOT NULL,
    data varchar(2048) NULL
);
GRANT SELECT, INSERT ON TABLE public.debezium_signal TO debezium_repl;

Inserts into the signal table are how a human requests a snapshot. Restrict INSERT to the people who are allowed to re-copy enrollment.

Connector configuration

jsonCopy
{
  "name": "enrollment-debezium",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "pg-primary",
    "database.port": "5432",
    "database.user": "debezium_repl",
    "database.password": "${file:/var/run/secrets/pg.properties:password}",
    "database.dbname": "techlearningpro",
    "topic.prefix": "enrollment",
    "plugin.name": "pgoutput",
    "slot.name": "enrollment_debezium",
    "publication.name": "enrollment_pub",
    "publication.autocreate.mode": "disabled",
    "table.include.list": "public.enrollment",
    "snapshot.mode": "initial",
    "tombstones.on.delete": "true",
    "heartbeat.interval.ms": "10000",
    "signal.data.collection": "public.debezium_signal",
    "tasks.max": "1",
    "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",
    "errors.tolerance": "none"
  }
}

tasks.max must be 1 for this connector. It reads one slot. Raising it does not parallelize a single PostgreSQL slot. topic.prefix=enrollment yields the topic enrollment.public.enrollment. heartbeat.interval.ms keeps the connector writing a heartbeat so the offset can move during quiet periods on the captured table, which lets the slot advance when the rest of the database is idle. Heartbeats do not replace a running task. A stopped task does not heartbeat.

errors.tolerance=none is deliberate. A tolerance of all on a source can skip a change the slot has already moved past, and there is no sink-style dead letter for "the WAL record we did not publish." A hard failure stops the task, the slot stays, WAL grows, and a human fixes the cause. That is louder than a hole in the cohort. Page on task FAILED and on retained WAL, which lesson 47 will put next to each other.

time.precision.mode and decimal modes are real Debezium settings. enrollment as defined here is keys, a status string, and timestamps. Pin time.precision.mode to the value your consumer tested if you include updated_at in the payload. A change of that mode rewrites the Connect schema and breaks a registry-backed subject if you later switch the converter to Avro. This lesson ships JSON so the fold can be read without a schema id. Lesson 43's rules apply the moment you convert the envelope to Avro. Do not auto-register the envelope from production.

There is no schema.history.internal.kafka.topic in this configuration. That topic is a MySQL binlog connector concern. Copying it from a MySQL tutorial onto PostgreSQL does not fix pgoutput and adds a topic you must now back up. PostgreSQL relation messages carry the shape pgoutput needs. If you add a MySQL source later, read that connector's page and create the history topic it requires, compacted, with enough retention to rebuild.

From envelope to the lesson 16 change

pythonCopy
def to_change(envelope: dict | None, key: dict | None) -> dict | None:
    """Map a Debezium record onto the change lesson 16 already folds.

    A tombstone has envelope is None. The key still identifies the row.
    """
    if envelope is None:
        if not key:
            raise ValueError("tombstone without a key")
        return {
            "lsn": None,
            "kind": "tombstone",
            "learner_id": key["learner_id"],
            "course_id": key["course_id"],
            "status": None,
        }
    op = envelope["op"]
    source = envelope["source"]
    if op == "d":
        row = envelope["before"]
        kind = "delete"
    elif op in ("c", "u", "r"):
        row = envelope["after"]
        kind = "snapshot" if op == "r" else "upsert"
    else:
        raise ValueError(f"unexpected op {op}")
    if row is None:
        raise ValueError(f"missing row image for op {op}")
    return {
        "lsn": source["lsn"],
        "kind": kind,
        "learner_id": row["learner_id"],
        "course_id": row["course_id"],
        "status": None if kind == "delete" else row["status"],
    }

The fold from lesson 16 applies upsert and snapshot as "this is the row" and delete as "remove the key," ignoring any change whose LSN is less than or equal to the LSN already applied for that key. A tombstone with lsn is None removes the key only when you do not already have a higher LSN. It must not bring a row back. Pass the mapped change into that fold. Do not rewrite the ordering rules inside the connector config.

Slot query

sqlCopy
SELECT slot_name,
       active,
       restart_lsn,
       confirmed_flush_lsn,
       pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS retained
FROM pg_replication_slots
WHERE slot_name = 'enrollment_debezium';

active false means the task is not connected. retained growing while active is false is the disk incident. confirmed_flush_lsn lagging pg_current_wal_lsn() while active is true means the connector is connected and behind. Page on retained bytes with a threshold that is a fraction of free disk, not on "any lag," because a minute of WAL is normal.

Incremental snapshot signal

sqlCopy
INSERT INTO public.debezium_signal (id, type, data)
VALUES (
    'snapshot-enrollment-2026-03-02',
    'execute-snapshot',
    '{"data-collections": ["public.enrollment"], "type": "incremental"}'
);

Run this when the cohort is wrong and the slot is healthy, and you need to re-read current rows. Do not drop the slot as the first response. A new slot starts a new initial snapshot only if the offset is also reset, and the gap between the old slot's death and the new snapshot is a window of lost deletes unless the snapshot replaces current state entirely. An incremental snapshot plus the LSN fold is the smaller hammer.

Practical Implementation or Design Walkthrough

Problem

Publish enrollment changes to Kafka and apply them to the cohort current-state table. Do not change the completion mart's delete behavior. Do not use the JDBC source.

Steps

1. Confirm wal_level, free replication slots, and a primary key on (learner_id, course_id). Create the role, the publication, and replica identity DEFAULT in a migration, not in the connector.
2. Install the Debezium PostgreSQL connector on the Connect plugin path from lesson 42. Restart workers so every worker can load the class. Submit enrollment-debezium with snapshot.mode=initial and errors.tolerance=none.
3. Watch the topic. You should see op=r events for existing rows, then silence until the next commit. Insert, update, and delete one lab enrollment. Expect c, u, and d, and a tombstone after the delete.
4. Run the Python mapping and the lesson 16 fold against those four events plus a replay of the same events. The second pass changes nothing. Apply the delete before the update by reversing a batch on purpose in a unit test, and show that sorting by LSN prevents the resurrection.
5. Stop the Connect task for a lab interval. Confirm active = false and that retained WAL grows if you generate unrelated WAL, or at least that restart_lsn does not advance. Start the task. Confirm it catches up without a second full snapshot. If it starts a snapshot, the offset was lost. That is a finding, not a success.
6. Point a second consumer at the same topic that builds nothing except a count of op=d per day for audit. Point the completion mart at the daily snapshot job, and show a deleted enrollment still present in yesterday's completion input.
7. Write the runbook line: retained WAL above the threshold pages the primary's owners and the connector's owners. Restarting the task is the mitigation when the task is down. Dropping the slot is a decision that accepts a new snapshot, not a disk cleanup you do at 02:00 without the fold's owner.

Validation queries

Cohort current state for lab key (l-1, course-a) is absent after the delete and present after a later insert with a higher LSN. A consumer rewind re-applies the topic and ends at the same row. purchase_line is absent from table.include.list. learning.progress has no slot.

Enterprise Real-World Scenarios

Unenrollment during a live cohort

Support deletes a test enrollment with SQL DELETE, the case lesson 16 opened with. pgoutput emits op=d with the primary key in before and in the Kafka key. The fold removes (learner_id, course_id) from the cohort table. The instructor view drops the seat within the poll-and-apply budget, which is dominated by the connector's heartbeat and the consumer lag, not by a fifteen-minute warehouse refresh. The completion mart for yesterday still shows the learner if they had completed work, because that mart reads the daily snapshot taken before the delete. Volume is the enrollment table, small next to progress events. The security constraint is the replication role: SELECT on enrollment and REPLICATION, password in a file on the worker, no superuser. The cost that matters is not the topic's bytes. It is WAL retained when the task is down during peak writes on the primary, including writes to tables you do not capture. The connector still has to read those WAL records to move the slot, which is fine when it is running and dangerous when it is not. The business decision is whether a seat is free. A stale "enrolled" is a room overfilled. A missed historical completion is a different incident, and this pipeline is not allowed to cause it.

Re-snapshot after a bad fold deploy

A fold deploy ignores op=d for an hour. The topic has the deletes. The cohort table does not. The slot is healthy. Rewinding the consumer group to the offsets from an hour ago and replaying with the fixed fold repairs the table, because the fold is idempotent on LSN and the topic retention covers the hour. An incremental snapshot is the alternative if you no longer trust the consumer's result and you want current rows re-emitted as op=r. You do not drop enrollment_debezium and you do not switch on the JDBC source "temporarily." The JDBC source would not see the deletes that are already gone from the table, and it would not repair the hour. Retention on enrollment.public.enrollment has to cover the rewind window you promise in the runbook. Lesson 39's retention setting is the control. This lesson's requirement is: longer than the maximum time you will take to notice a bad fold and replay, and long enough to rebuild current state from snapshot events plus subsequent changes.

Best Practices

• One slot, one connector, tasks.max=1, a publication that lists enrollment and not ALL TABLES.
• publication.autocreate.mode=disabled. Migrations create publications.
• REPLICA IDENTITY DEFAULT while the fold only needs the key on delete. Write down the day you switch to FULL and why.
• errors.tolerance=none on this source. A skipped WAL event is a missing unenrollment.
• Heartbeats on, so a quiet enrollment table does not pin the slot while the task is actually up.
• Topic key equals the primary key. Apply in LSN order per key. Replay is a no-op on the same LSN.
• Tombstones may follow deletes. The fold's source of truth for "this was a delete" is op=d.
• Monitor active and retained bytes. A down task is a primary-disk page, not only a data-delay page.
• Reset offsets and slots only as a pair, with a written snapshot plan.
• Keep a second consumer for history off this delete stream. Daily snapshots feed completion.
• Do not put progress or purchase lines on this slot.
• Pin the Debezium version and read its PostgreSQL page for snapshot.mode names before an upgrade. A rename of a property is a deploy, not a surprise.
• Store the connector password through the config provider from lesson 42.

Common Mistakes and Anti-Patterns

JDBC mode "until Debezium is ready." Deletes in that interval never hit Kafka. The cohort is wrong in the direction that overfills a room. If the product needs deletes, the first production connector is the log reader.

publication.autocreate.mode left at a value that publishes every table. The slot now carries tables you did not review, including ones with personal data. The topic list grows. Set disabled and list public.enrollment.

REPLICA IDENTITY NOTHING left from a replica experiment. Delete events have no key the fold can apply. Unenrollments vanish. Set DEFAULT and confirm the primary key.

Dropping the slot to free disk, as a reflex. The next start snapshots or begins at a new LSN. Changes in the gap are gone unless a snapshot replaces the whole current state and you accept that transactions in flight need a story. Restart the task. Expand disk if you must. Drop the slot with the fold's owner.

Deleting Connect offsets as a cleanup. The connector's memory of the LSN is gone. Combined with when_needed or initial, you can snapshot during class. Treat offset deletion like a rebuild plan.

Applying events in consumer-arrival order across partitions with a random key. Two workers interleave one enrollment key and an older update lands last. Key by (learner_id, course_id) and apply LSN order per key.

One warehouse job that applies op=d to every table it sees. Friday's unenrollment deletes Wednesday's completion. Split the consumers.

Compacting the only change topic and promising transition history. Compaction keeps the latest record per key. Intermediate statuses disappear. If you need both current state and history, retention without compaction is the history, and current state is the fold's table.

Copying the MySQL schema-history topic into the PostgreSQL config and calling the setup done. pgoutput does not use that topic. You still need the publication, the slot, and replica identity.

Superuser for debezium_repl. The connector needs replication and select, not the right to drop databases. Extra privilege turns a connector bug into a primary incident.

Troubleshooting and Design Review

Symptom: The cohort still shows a learner who was deleted in PostgreSQL. The topic has no op=d.

Likely cause: the delete happened before the slot existed and the snapshot already omitted the row, so the cohort was loaded from somewhere else; or the table is not in the publication; or replica identity cannot identify the row and the event was skipped; or you are looking at a JDBC topic.

Investigation: publication definition, REPLICA IDENTITY, the slot's existence at the time of the delete, and the topic name. Search for the key.

Corrective action: if the publication was wrong, fix it and incremental-snapshot. If the cohort was loaded from a poll, rebuild current state from op=r plus subsequent changes. Do not hand-delete one learner in the cohort without an LSN, or the next replay will put them back.

Symptom: Primary disk is climbing. enrollment_debezium shows active = false.

Likely cause: the task is down and the slot is retaining WAL. Unrelated write traffic makes it worse. The slot does not care that you only capture one table when it is not confirming.

Investigation: Connect task status and the retained-bytes query. Worker logs for the exception that killed the task.

Corrective action: fix the exception and start the task. Confirm active returns to true and retained bytes stop growing. If disk is critical and the task cannot start, the database owner decides whether to drop the slot. That decision includes a rebuild of the cohort. Do not make it the connector on-call's private choice.

Symptom: After a restart, the topic receives another full set of op=r events.

Likely cause: the Connect offset was missing, so snapshot.mode=initial or when_needed snapshotted again. The slot may be new as well.

Investigation: GET /connectors/enrollment-debezium/offsets and pg_replication_slots.

Corrective action: if the fold is idempotent, the extra snapshot is noisy and safe. Find who deleted the offset. If the fold duplicated rows, stop the consumer and fix the LSN check before another replay. Pin snapshot.mode so an offset loss does not choose a mode you have not rehearsed.

Symptom: Status updates appear out of order in the cohort table.

Likely cause: the apply is not ordered by LSN per key. A parallel consumer or a batch that sorted by ts_ms instead of lsn resurrected a row. Connector processing time is not the order.

Investigation: the stored LSN for the key versus the event that "won."

Corrective action: sort and gate on source.lsn. Rewind the consumer. Do not sort on ts_ms. Two commits can share a millisecond. They cannot share an LSN order.

Symptom: The completion count for Wednesday dropped after Friday's account-deletion job.

Likely cause: a consumer applied op=d to the historical fact. The connector did what it should. The subscriber did not.

Investigation: which consumer group wrote the mart, and whether it treats d as a current-state removal only.

Corrective action: restore the mart from the daily snapshot path. Remove the CDC subscription from that job. Add a test that a delete event leaves Wednesday's completion row in place.

A design review asks: Which table is in the publication? What is the slot name, and who is paged when it is inactive? Where is the Connect offset stored? What is the topic key? What does the fold do with op=d, with a tombstone, and with a repeated LSN? Which consumer is forbidden to delete history? What is the rewind retention? Why is progress not on this slot?

Interview Questions and Answers

Q1. How does the Debezium PostgreSQL connector implement the snapshot handshake from the CDC fundamentals lesson?

With snapshot.mode=initial it takes a consistent snapshot, emits one op=r event per current row with an LSN in source, then streams later commits as c, u, and d. The Connect offset stores how far that process got so a restart does not snapshot again. That is the coordinated snapshot plus slot from lesson 16, inside a Connect source. I still fold in LSN order per key. The connector does not maintain the cohort table for me.

Q2. Why is pgoutput the plugin in this design?

It ships with PostgreSQL and it is the plugin logical replication already uses, so the primary does not need a separate decoding library. The publication lists the tables it will emit. I set publication.autocreate.mode=disabled and create enrollment_pub in a migration so the connector cannot widen the set. decoderbufs would be an extra extension I do not need for this table.

Q3. What do op values r, c, u, and d mean for the cohort fold?

r is a snapshot read and c and u are inserts and updates. All three present an after image that becomes the current row. d is a delete and presents the key, which the fold removes from current state. t would be a truncate and I do not subscribe casually, because it means every key is gone. I map the envelope into the change record lesson 16 already knows how to apply. I do not invent a second ordering field.

Q4. Why is replica identity DEFAULT enough here, and when would you switch to FULL?

DEFAULT logs the primary key. The fold needs (learner_id, course_id) on a delete and the new status on an upsert. It does not need the previous status. FULL logs the entire old row, which I would turn on only for a consumer that computes deltas from the before image, and I would expect more WAL. NOTHING would make deletes unapplyable. Changing the identity later does not reconstruct images for WAL that is already gone.

Q5. What is the tombstone for, and why is it not sufficient as the only delete signal?

Debezium's default tombstones.on.delete=true writes a null-valued record with the same key after the op=d event, so log compaction can remove the key. The op=d value carries source.lsn and the before image. The tombstone does not carry that LSN in the value, because the value is null. I apply op=d in the fold. I treat a lone tombstone as a key removal only as a recovery path, and I never let it recreate a row.

Q6. A task has been down for an hour. What is the primary risk, and what is the wrong first fix?

The slot is inactive and PostgreSQL retains WAL from the restart LSN. Unrelated traffic on the primary makes the retained bytes grow until the disk fills and the primary stops accepting writes. The cohort being stale is the smaller problem. The wrong first fix is to drop the slot to free disk without a rebuild plan. I start the task if I can, watch retained bytes fall, and I involve the database owner before any slot drop. Heartbeats do not help while the process is dead.

Q7. Name the three cursors and a mistake that confuses two of them.

The replication slot in PostgreSQL, the Connect source offset in connect-offsets, and the cohort consumer's group offset. A mistake is to rewind the consumer and expect the slot to rewind, or to delete the Connect offset and expect the slot to protect you from a new snapshot. The consumer rewind is safe only because the fold is idempotent on LSN. The offset deletion can trigger snapshot.mode to copy the table again. I reset them only in pairs, on purpose.

Q8. Why does Connect exactly.once.source.support not retire the LSN check?

When the worker and the connector both support it, the produce to Kafka and the Connect offset commit can be one transaction. The cohort table is updated by a different consumer, later, in a different database. A crash between those writes redelivers the event. Two Kafka records are still two deliveries of one LSN. The fold's "same LSN is a no-op" is the cohort's correctness. I turn on source exactly-once only after the Debezium version's notes say the PostgreSQL connector supports it, and I keep the LSN check either way.

Q9. Why must the completion mart ignore this delete stream?

The mart's Wednesday row is a historical fact. Friday's op=d means the learner is not in the cohort now. Applying the delete to the mart erases a completion that was true on Wednesday. Lesson 16 split the products. The connector gives both consumers the same topic. I subscribe the cohort fold to that topic and I feed the mart from the daily snapshot. A shared "CDC apply" framework without a per-product delete rule will eventually pick the wrong one.

Q10. How is this different from the JDBC source on purchase lines?

The JDBC source polls line_seq and cannot see updates or deletes. Purchase lines are insert-only, so that gap is the same as the business rule. Enrollment rows are deleted. A poll cannot emit a key that is gone. Debezium reads the WAL through pgoutput and emits op=d. I do not put enrollment on the JDBC connector while Debezium is "in progress," because the deletes during that period are the ones support just issued.

Q11. What do you do when the fold was wrong for an hour but the topic is right?

I fix the fold, rewind the cohort consumer to before the bad hour, and replay. Idempotency on LSN makes the good events no-ops and applies the deletes the bad deploy skipped. I need topic retention that covers the hour. If the topic is gone, I use an incremental snapshot signal to re-emit current rows as op=r and I accept that intermediate transitions in the gap are lost. I do not switch the table to a query watermark to "fill in" the hour.

Hands-On Exercises

Exercise: An enrollment delete that reaches the cohort and stops there

Problem statement. Run the Debezium PostgreSQL connector against a lab enrollment table. Show a delete, a replay that does not resurrect, and a stopped task that retains WAL. Show that a completion-style row is untouched by op=d.

Tasks.

1. Create enrollment with primary key (learner_id, course_id), wal_level=logical, the publication, the replication role, and REPLICA IDENTITY DEFAULT.
2. Submit the connector with plugin.name=pgoutput, snapshot.mode=initial, tasks.max=1, and errors.tolerance=none. Capture one op=r event.
3. Insert, update, and delete a row. Show c, u, d, and the tombstone. Map them with to_change.
4. Apply the mapped events with the lesson 16 LSN rule twice. The second apply is a no-op. Apply them out of LSN order in a test and show the gate that blocks resurrection.
5. Stop the task. Record active and retained WAL from pg_replication_slots. Start the task and show the slot moving again.
6. Write one sentence on why you will not drop the slot as a disk fix, and one sentence on why the completion mart does not consume op=d.
7. Confirm purchase_line is not in table.include.list.

Expected deliverables. The connector JSON, the four decoded operations, the fold test, the slot query output while stopped, and the two sentences.

Evaluation rubric.

You pass when a SQL DELETE is visible as op=d, the fold drops the key, a second replay leaves it dropped, and the slot query shows active = false while the task is stopped.

Lesson Summary

Debezium's PostgreSQL connector is the Connect source that turns lesson 16's log capture into a topic. pgoutput, a narrow publication, and snapshot.mode=initial implement the snapshot-then-stream handshake. source.lsn remains the order key. op=d removes cohort current state and must not remove historical completions. Replica identity DEFAULT is enough while deletes only need the primary key. The slot, the Connect offset, and the consumer group are three cursors. A stopped task retains WAL on the primary, which is a disk incident, not only a stale dashboard. The JDBC source stays on insert-only purchase lines. Progress events stay off the slot. The fold's idempotency is still the cohort's correctness under redelivery.

Real-Time Data Quality and Monitoring is next. The slot bytes, the sink lag, the dead-letter topic, and the tile's lateness become numbers with owners.

Further Reading

• Debezium PostgreSQL connector — plugin.name, topic.prefix, snapshot.mode, publication settings, the envelope, and tombstones. Check the page for the version you deploy. Property names have moved between major releases.
• Debezium incremental snapshots — the signal table and execute-snapshot.
• PostgreSQL logical decoding — slots, WAL retention, and why an inactive slot fills disk.
• PostgreSQL publications — what enrollment_pub actually limits.
• Change Data Capture Fundamentals — the fold this connector must not replace.
• Kafka Connect — the worker that hosts the plugin, and the offset topic that stores the LSN.
• /learn/kafka — the topic the envelope is written to, and the consumer offsets that form the third cursor.