Change data capture and the outbox, lab

What it is and why it exists

A service that changes its database and also publishes an event to Kafka has a problem. If it writes twice, one write can succeed while the other fails. Module 08 explains this "dual write" problem and names two answers: change data capture and the outbox pattern. See module 08 for the theory, Kafka Connect, dead-letter topics and the schema registry. This module does not repeat them.

Change data capture means a program reads the database's own change log and publishes each insert, update and delete as a Kafka record. The tool here is Debezium, a set of source connectors for Kafka Connect. The database is PostgreSQL 16. You run both and read the records they produce.

The lab shows four things: what a change record looks like, how an outbox keeps rolled-back work out of Kafka, how the database behaves while a connector is stopped, and one odd result about lag that you should expect.

Mental model

Think of two ways to learn what happened in a database.

Words used in this module

• Write-ahead log (WAL). The file in which PostgreSQL records every change before it applies it. Capture tools read it.
• Logical decoding. A PostgreSQL feature that turns the WAL into row-level changes. It needs the server setting wal_level=logical.
• Replication slot. A bookmark inside PostgreSQL that remembers how far a reader has confirmed. PostgreSQL keeps all log after that point.
• Publication. A named list of tables whose changes are sent to readers.
• LSN. Log sequence number: a position in the WAL. Each change record carries one.
• Snapshot. When a connector first starts it reads the rows that already exist. Those records have operation code r.

Use the tiles at the top of the page. Pick each operation and note which of before and after is empty.

How it works internally

The path of one change

1. The application commits a transaction. PostgreSQL writes it to the WAL.
2. The replication slot lets the connector read decoded changes. The lab uses the output plugin pgoutput.
3. The connector sends each change into Kafka. For the table public.orders and topic prefix shop, the topic is shop.public.orders.
4. Later, the connector tells PostgreSQL how far it has safely delivered. Only then can PostgreSQL discard older log.

The change record

The record key is the table's primary key, so all changes to one row go to the same partition in order. The value holds an operation code, a before image, an after image and a source block.

All four rows of this table were seen in the lab, with REPLICA IDENTITY FULL set on the table. After a delete record, the lab topic also held a second record with the same key and a null value. That is a tombstone: for a compacted topic it marks the key for removal.

What the before image holds without REPLICA IDENTITY FULL was not tested. The usual claim is that it is empty or holds only the key columns. unverified

How the outbox is routed

The second connector in the lab uses a transform, a small step that rewrites each record. The transform is Debezium's EventRouter. It turns an outbox row into a plain event. In the lab, the routing went by the column aggregatetype: rows with order landed in the topic outbox.event.order. The message key was the aggregateid column. The value was the payload column. Each message had a header id holding the row's uuid.

Where the event type (OrderCreated or OrderPaid) goes was not visible in the consumer output, and I did not check it. It may be a header. unverified

Why a stopped connector costs disk

A slot holds a position. PostgreSQL may not delete log after that position. While a connector runs, the position advances. While it is stopped, the slot is inactive and the position stays still, so the log grows. Part C of the lab measures this.

Failure scenarios

Hands-on lab

Step 1. Start the stack

cd ~/Desktop/kafka-knowledge-base
docker compose --profile cdc up -d
k() { docker exec -i kafka-1 /opt/kafka/bin/"$@"; }
P() { docker exec -i pg psql -U postgres -q -A -t -c "$1"; }
C=http://localhost:8087

The service pg is PostgreSQL started with -c wal_level=logical. The service cdc-connect publishes its REST interface on host port 8087.

Step 2. Create the tables

P "CREATE TABLE orders (id int PRIMARY KEY, customer text NOT NULL, amount numeric(10,2) NOT NULL, status text NOT NULL)"
P "CREATE TABLE outbox (id uuid PRIMARY KEY DEFAULT gen_random_uuid(), aggregatetype text NOT NULL, aggregateid text NOT NULL, type text NOT NULL, payload jsonb NOT NULL, created timestamptz NOT NULL DEFAULT now())"
P "ALTER TABLE orders REPLICA IDENTITY FULL"
P "INSERT INTO orders VALUES (1,'ana',40.00,'created')"

One row exists before any connector. The connector will snapshot it.

Step 3. Part A: register the connector for the table

curl -s -X POST $C/connectors -H 'Content-Type: application/json' -d '{
  "name": "orders-cdc",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "plugin.name": "pgoutput",
    "database.hostname": "pg", "database.port": "5432",
    "database.user": "postgres", "database.password": "lab",
    "database.dbname": "postgres",
    "topic.prefix": "shop",
    "table.include.list": "public.orders",
    "slot.name": "orders_slot",
    "publication.name": "orders_pub",
    "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"
  }}'
# wait about 12 seconds
curl -s $C/connectors/orders-cdc/status

After about 12 seconds the connector was RUNNING and task 0 was RUNNING.

Step 4. Change the table and read the topic

P "INSERT INTO orders VALUES (2,'ben',15.50,'created')"
P "UPDATE orders SET status='paid' WHERE id=1"
P "DELETE FROM orders WHERE id=2"
k kafka-console-consumer.sh --bootstrap-server kafka-1:19092 --topic shop.public.orders --from-beginning --formatter-property print.key=true --timeout-ms 7000

key | op | before | after | source
{"id":1} | r | null | {"id":1,"customer":"ana","amount":"D6A=","status":"created"} | snapshot=last lsn=22182616
{"id":2} | c | null | {"id":2,"customer":"ben","amount":"Bg4=","status":"created"} | snapshot=false lsn=22460936
{"id":1} | u | {... status created} | {... status paid} | snapshot=false lsn=22461128
{"id":2} | d | {... ben ... created} | null | snapshot=false lsn=22461288
{"id":2} | (value null: a tombstone, after the delete record)

Read it in order. The first row is the snapshot of the row that existed before the connector. The delete is followed by a tombstone for the same key. The amount shows as base64 text, as described in the failure scenarios.

Step 5. Part B: register the outbox connector

curl -s -X POST $C/connectors -H 'Content-Type: application/json' -d '{
  "name": "outbox",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "plugin.name": "pgoutput",
    "database.hostname": "pg", "database.port": "5432",
    "database.user": "postgres", "database.password": "lab",
    "database.dbname": "postgres",
    "topic.prefix": "shopout",
    "table.include.list": "public.outbox",
    "slot.name": "outbox_slot",
    "publication.name": "outbox_pub",
    "transforms": "outbox",
    "transforms.outbox.type": "io.debezium.transforms.outbox.EventRouter",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false"
  }}'

The connector reached RUNNING.

Step 6. Three transactions

P "BEGIN; INSERT INTO orders VALUES (3,'cy',99.00,'created'); INSERT INTO outbox (aggregatetype,aggregateid,type,payload) VALUES ('order','3','OrderCreated','{\"orderId\":3,\"amount\":99.00}'); COMMIT;"
P "BEGIN; INSERT INTO orders VALUES (4,'di',10.00,'created'); INSERT INTO outbox (aggregatetype,aggregateid,type,payload) VALUES ('order','4','OrderCreated','{\"orderId\":4,\"amount\":10.00}'); ROLLBACK;"
P "BEGIN; UPDATE orders SET status='paid' WHERE id=3; INSERT INTO outbox (aggregatetype,aggregateid,type,payload) VALUES ('order','3','OrderPaid','{\"orderId\":3}'); COMMIT;"
k kafka-topics.sh --bootstrap-server kafka-1:19092 --list | grep -E 'outbox|shop'
k kafka-console-consumer.sh --bootstrap-server kafka-1:19092 --topic outbox.event.order --from-beginning --formatter-property print.key=true --property print.headers=true --timeout-ms 7000
P "SELECT id, status FROM orders ORDER BY id"
P "SELECT count(*) FROM outbox"

topics listed (new ones): outbox.event.order, shop.public.orders

outbox.event.order: exactly 2 messages, key "3" for both
  value 1: "{\"amount\": 99.00, \"orderId\": 3}"
  value 2: "{\"orderId\": 3}"
  each with headers including id:<uuid> and __debezium.context.*
  nothing for order 4 (rolled back)

orders rows: 1|paid and 3|paid
outbox row count: 2

The rolled-back transaction produced nothing in Kafka. The payload arrived as an escaped JSON string. Both messages for order 3 share a key, so they stay in order on one partition. The flags in the console consumer line that print headers are my addition for this page; the recorded output showed headers, but I did not keep the exact flag. unverified

Step 7. Part C: stop a connector and watch the slot

SLOTS="SELECT slot_name, active, pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn) FROM pg_replication_slots"
P "$SLOTS"
curl -s -o /dev/null -w '%{http_code}\n' -X PUT $C/connectors/orders-cdc/stop
curl -s $C/connectors/orders-cdc/status
P "INSERT INTO orders SELECT g,'bulk',1.00,'created' FROM generate_series(1000,60999) g"
P "$SLOTS"

before:  orders_slot  active=t  23 kB behind
         outbox_slot  active=t  24 kB behind

stop:    HTTP 204; status STOPPED; tasks []

after 60,000 inserts:
         orders_slot  active=f  8607 kB behind
         outbox_slot  active=t  8607 kB behind

The second slot is active, but its table saw no change. It is still 8607 kB behind.

Step 8. Resume and measure

curl -s -o /dev/null -w '%{http_code}\n' -X PUT $C/connectors/orders-cdc/resume
# run the slot query again over 40 seconds
k kafka-get-offsets.sh --bootstrap-server kafka-1:19092 --topic shop.public.orders

resume: HTTP 202
bytes behind over 40 s: 8813272 -> 8852600 (about 8.8 MB, not falling)
topic shop.public.orders end offset: 60007  (7 + 60,000)

All data reached Kafka, yet the slot did not move within 40 seconds. The end offset command shown above is my wording for how to read it; the recorded value is the one printed. unverified

Step 9. Shut down

docker compose --profile cdc down

Common mistakes and misconceptions

Staff-level questions

Self-test