In this lab, you run a Debezium PostgreSQL source connector on one distributed Connect worker, follow committed row changes from public.stock into a Kafka topic, and watch the worker resume from its saved LSN after a restart and after a database outage. You then add the transactional outbox: one PostgreSQL transaction writes an order and its OrderPlaced event, and the Outbox Event Router publishes that event to events.orders with the order ID as the key and the event ID as a header.

The stock table starts with two rows, and each phase applies a known set of changes. One order, ord-1042, produces one outbox row with event ID 8b147168-9c23-4af8-a4d9-5bc5c02b3f91. The counts below describe a fresh, successful run. Check saved source progress and explain repeated deliveries if a checkpoint flush or recovery attempt differs. In Phase 6, a new connector deliberately snapshots the same outbox row again.

Environment: E4, the change-data-capture stack. Docker on your host runs three containers on one compose network: kafka (one combined broker and controller), postgres (PostgreSQL 17 with wal_level=logical), and connect (one Debezium Connect worker in distributed mode). Console consumers run inside the kafka container against kafka:9092. psql runs inside the postgres container through docker exec. curl runs on your host against localhost:8083. Open three terminals: Terminal A for the stock consumer, Terminal B for the events.orders consumer, and Terminal C for everything else.

Create:

Save this as docker-compose.yml. Replication factors are 1 because there is one broker. The worker's three internal topics are the configuration, source-offset, and status stores.

From kafka-labs/cdc/ in Terminal C:

The worker needs a minute to create its internal topics and join its group. Poll the root endpoint until it answers, then list connectors, plugins, and topics, and confirm the database prerequisite for logical decoding:

Save all outputs as observations/01-worker.txt.

What you should see: the root endpoint returns a JSON object whose kafka_cluster_id is 4L6g3nShT-eMCtK--X86sw, the connector list is [], the plugin list contains io.debezium.connector.postgresql.PostgresConnector, the topic list contains connect-configs, connect-offsets, and connect-status, and wal_level is logical. The worker has a place to store configuration and checkpoints, and no integration yet.

Create the table with two rows before the connector exists, so its initial snapshot has something to read. REPLICA IDENTITY FULL makes old column values available for updates and deletes. Save this as sql/01-stock.sql:

Save this as connectors/inventory-source.json. The slot and publication names are explicit because a second connector joins the same database in Phase 5. Both converters use schemas.enable=false so the envelope prints without a schema wrapper. errors.max.retries is 0 so Phase 4 surfaces a database outage as a failed task instead of a retry loop.

Register it and check its status:

In Terminal A, start a consumer on the table topic and leave it running for the rest of the lab:

Wait until Terminal A shows both snapshot records at versions 85 and 3. Then apply one insert, one update, and one delete. Save this as sql/02-stock-changes.sql and run it like the first file:

Copy every Terminal A line into observations/02-stock-events.txt, and under each envelope note its op, whether before and after are present, and source.lsn.

What you should see: the connector and task 0 both report RUNNING. A clean run prints the six records below. Snapshot order can vary; use each record’s key and operation to identify it.

The tombstone follows the delete event. Record the streamed changes’ source.lsn values and compare their positions; a tombstone has no change envelope or LSN. Snapshot reads describe existing rows, not new inserts.

The source checkpoint lives in the worker's offset topic, not in __consumer_offsets. The worker flushes source offsets about once a minute by default, so repeat this until the latest checkpoint for inventory-source reaches the last streamed Phase 2 change's source.lsn. Ignore the tombstone when choosing that change, because it has no LSN:

Save the output as observations/03-checkpoint.txt and write the lsn value next to the source.lsn of the last Phase 2 event.

Stop the worker gracefully, which lets it flush its final checkpoint, then make three changes while no worker is running, and check the slot from the database side:

Append the slot output and everything new from Terminal A to observations/03-checkpoint.txt.

What you should see: the offset record's key names inventory-source and the server inventory; its value is a JSON object with LSN fields, and lsn is at or beyond the last Phase 2 source.lsn. While the worker is stopped, the slot row shows active as f and Terminal A prints nothing. After the restart, Terminal A prints exactly three new envelopes, u for sku-1042 at version 87, c for sku-1045, and u for sku-1044, with no Phase 2 replay if the final checkpoint flush completed successfully. If records repeat, preserve that evidence and compare the checkpoint with the source positions. The worker read its configuration from connect-configs, its position from connect-offsets, and PostgreSQL had retained the WAL the slot needed.

Now break the source instead of the worker. Stop the database while the task is streaming and poll the status until it changes:

Save the failed status as observations/04-failure.txt. Bring the database back, check the status again before touching the task, then restart the task and make one more change:

Append every status response and the new Terminal A line to observations/04-failure.txt.

What you should see: after the database stops, the connector stays RUNNING while task 0 reports FAILED with a trace naming a connection failure to postgres. The database coming back changes nothing: the task is still FAILED, because the base runtime does not restart a failed task on its own. The restart request returns 204 No Content, the next status shows task 0 RUNNING, and Terminal A prints one u envelope at version 88. The slot survived the outage, so the task continued from its checkpoint instead of snapshotting again.

Save this as sql/03-outbox-schema.sql and run it:

Register the second connector before any order exists, so the event you observe comes from streaming rather than a snapshot. Save this as connectors/outbox-source.json; it uses its own slot and publication, and the transform and predicate block from the outbox material.

Check docker compose logs connect for this outbox connector's successful empty-snapshot completion and transition to streaming before inserting the order. A RUNNING task status alone does not prove the snapshot has finished. In Terminal B, start a consumer that prints headers, keys, partitions, and offsets:

Now act as the order service. Both inserts sit inside one transaction on one connection. Save this as sql/04-place-order.sql and run it:

Save the Terminal B output and a fresh topic listing as observations/05-outbox-events.txt.

What you should see: Terminal B prints one record. Its headers include id with 8b147168-9c23-4af8-a4d9-5bc5c02b3f91 and eventType with OrderPlaced. Its key is the plain string ord-1042, from aggregateid, and its value is the payload as a JSON object with orderId and aggregateVersion, not an escaped string and not a before/after envelope. The topic listing contains events.orders and no ordersdb.public.outbox_event, because the router renamed the record before Connect published it. Terminal A printed nothing: the order tables are outside the first connector's table list.

The outbox row is still in the table, and a connector with a new name has no checkpoint, so it snapshots the table and routes the row again. Copy outbox-source.json to connectors/outbox-source-replay.json and change three values: name to outbox-source-replay, slot.name to outbox_replay_slot, and publication.name to outbox_replay_pub. Remove the original connector and register the copy. Wait until Terminal B shows the replayed event, then read the whole topic in Terminal C and list the slots:

Save both as observations/06-replay.txt.

What you should see: Terminal B prints a second record with the same id header, key, and payload. The full read shows two records in events.orders at different offsets, and nothing in Kafka marks the second as a duplicate; only the id header ties them together. The slot listing shows inventory_slot and outbox_replay_slot active and outbox_slot inactive but present, because deleting a connector does not drop its slot, and that leftover slot retains WAL until someone removes it.

Stop the consumers in Terminals A and B with Ctrl-C, then remove the stack and its volumes:

Verify that nothing remains:

The listing should show no containers. The images stay downloaded. Keep the kafka-labs/cdc/ folder; it is your submission.

One database produced two different kinds of Kafka records. The stock connector published raw change envelopes, and their op codes, before and after fields, and LSNs let you tell a snapshot read from a streamed insert. The connect-offsets record was the source checkpoint: a database position, not a Kafka offset, stored in Kafka so a restarted worker could resume from the slot instead of snapshotting again. Stopping PostgreSQL showed the other side of that dependency, a task that failed and stayed failed until you fixed the cause and asked for a restart.

The outbox phases turn a row insert into an OrderPlaced event with a deliberate topic, key, and headers. The database transaction commits the order and its outgoing event together. A new connector then snapshots the same outbox row and publishes its event ID at another Kafka offset. Kafka stores those publications as separate records; consumers use the stable event ID to recognize the repeated business event.