"Plugging Into the WAL" — Kafka Connect, Debezium and Postgres, End to End

Story Opening

Zoya’s first connector came from a blog post written in 2021. It connected as the database owner, set database.server.name, and left plugin.name unset. Connect answered 400: The 'topic.prefix' value is invalid: A value is required. She renamed the property, got a cheerful 201, and seconds later the task was FAILED on the decoderbufs default she’d read about in Part 1’s gotchas and forgotten.

“You said Connect is what I’ll be operating,” she said. “So teach me Connect.”

Tomás pulled up a chair. “Its offsets, its topics, its rebalances, and its idea of ‘running’, which isn’t yours.” He pointed at her config. “And it connects as the owner of every table. We’ll fix that first.”

By Friday she wanted three things: a connector that could only read what it needed, events she understood down to the last header, and a tracking-projection that didn’t care if Connect fell over mid-stream. That last one turned out to be the real work.

The Quick Map

Conceptual Deep-Dive

Debezium is a Kafka Connect source connector

Most of what surprises people about Debezium is really Kafka Connect behaviour. Connect is a cluster of workers (JVMs). You submit a connector configuration through its REST API; the connector splits the work into tasks; a task’s poll() returns SourceRecords; the worker runs them through transformations (SMTs), then a converter, and hands them to its own embedded producer. Every so often the worker stores the task’s source offsets in a Kafka topic, so a restarted task knows where to resume.

Run it in distributed mode, as here: workers that share a group.id form one Connect cluster and coordinate through the brokers, much like a consumer group. Standalone mode keeps offsets in a local file and has no failover, so it’s for experiments only. When a worker joins or leaves, the cluster rebalances with incremental cooperative rebalancing (connect.protocol = sessioned in the lab’s worker). Only the tasks that have to move are stopped and restarted elsewhere. For a Debezium task, moving means: stop, flush offsets if it can, reconnect to the slot from another worker, resume from the stored offset. A worker that stops cleanly hands its tasks over within seconds. A worker that simply vanishes is different: the group leader holds its tasks for scheduled.rebalance.max.delay.ms, 300000 ms (five minutes) by default, in case it comes back. Part 3 measures both on a two-worker cluster. (Connect runs sink connectors too: Debezium’s JDBC sink can write these events straight into another database, which this series doesn’t need.)

Why the PostgreSQL connector runs one task

tasks.max is ignored. The docs put it plainly: “The PostgreSQL connector always uses a single task and therefore does not use this value.” A replication slot is one ordered stream with one reader, so a second task would have nothing to read. To scale, split tables across connectors, each with its own slot (and usually its own publication and topic.prefix). Two connectors must never share a slot. Troubleshooting below shows how quietly that fails: the task never even turns FAILED.

Where to flatten the envelope

Debezium’s before/after/source/op envelope is verbose, and the ExtractNewRecordState SMT can flatten it to just the new row. Whether to do that on the connector is a real trade-off:

This series keeps the raw envelope on dispatchly.public.shipments and lets each consumer pick what it needs. If many simple consumers want flat rows, a second connector or a stream processor can produce a flattened topic from it. Part 3 adds the third option: a published contract via the outbox, instead of either.

Correctness lives in the consumer

Part 1 established that delivery is at-least-once. This part makes that concrete: you’ll kill a Connect worker and watch every event since its last offset flush arrive twice. The projection’s job is to make that boring. It follows three rules:

1. Apply a change only if it’s newer than what you have. WHERE view.version < incoming.version turns duplicates and out-of-order replays into no-ops, if every write raises version. A trigger makes sure of that (step 1).
2. Treat a delete as terminal. Keep a marker row instead of deleting, so a replayed older u or r event can’t resurrect the shipment. The price: an id that genuinely comes back (an explicit id reused, a primary key changed back) never reappears in the view.
3. Never parse what you don’t need. A projection that only reads status, hub_id and eta survives new columns and odd encodings elsewhere in the row.

Why version and not source.lsn? Both work as a per-row guard. Spring Data JDBC’s @Version raises version on every save(), but a DBA’s hot-fix wouldn’t, and the view would stay wrong until that row next changed, hence the trigger. source.lsn needs no discipline from writers (row locks serialise them), but it belongs to one database’s WAL: it can go backwards after a restore and starts over on a new cluster. version is a business fact that survives both.

Technical Explanation

Connect’s internal topics

The worker stores everything in three compacted topics, named in its configuration, and creates them itself on first start (kafka-topics.sh --describe, trimmed):

connect-configs   partitions 1   cleanup.policy=compact
connect-offsets   partitions 25  cleanup.policy=compact
connect-status    partitions 5   cleanup.policy=compact

connect-offsets is the one to respect. Here is the record Debezium’s PostgreSQL connector stored right after its initial snapshot, read straight off the topic with print.key=true:

["shipments-cdc",{"server":"dispatchly"}]  {"lsn":33900608,"txId":775,"ts_usec":1792693551826194}

The key is the connector name plus Debezium’s source partition (server = topic.prefix); the value is the LSN to resume from. Once streaming starts, the offset grows lsn_proc, lsn_commit and messageType fields. Delete this topic, or rename the connector, and the connector has no idea where it was: it snapshots again.

Converters decide what consumers see

The worker defaults to JsonConverter with schemas embedded. With value.converter.schemas.enable=true, each event becomes {"schema": {…}, "payload": {…}}. In the lab, one update event was 3,862 bytes that way, over 80% of it schema: field types plus Debezium’s logical type names such as io.debezium.time.ZonedTimestamp for timestamptz. This series sets schemas.enable=false at the connector level for readable payloads. Part 3 moves to Avro with a schema registry, which keeps the types and drops the bulk. Producer and consumer must agree; the troubleshooting table shows what happens when they don’t.

SMTs and predicates

Transformations run inside the worker, after the task and before the converter, in the order listed in transforms. A predicate limits a transformation to matching records. The built-ins are TopicNameMatches, HasHeaderKey and RecordIsTombstone, and negate=true inverts one. Debezium connectors also emit records you don’t want transformed, such as heartbeats on __debezium-heartbeat.<prefix>, so pinning an SMT to the data topics is a good habit. Step 4 does exactly that, and Part 3’s outbox router depends on it.

Plugins and how Connect finds them

Connect loads each directory on plugin.path with its own class loader, so two connectors can bring conflicting library versions. The Debezium image ships every Debezium connector under /kafka/connect (plugin.path=[/kafka/connect] in the worker’s log). On a plain apache/kafka image you’d download debezium-connector-postgres-3.7.0.Final-plugin.tar.gz from Maven Central, unpack it into your plugin path, and run connect-distributed.sh. That is how Debezium’s own “plain Kafka container images” example does it. Kafka 4.3’s plugin.discovery defaults to hybrid_warn (scan, and warn about plugins missing ServiceLoader manifests). GET /connector-plugins is the truth either way.

Topic auto-creation for source connectors

The lab’s broker runs with auto.create.topics.enable=false, so nothing appears by accident. The connector’s topic.creation.default.replication.factor and topic.creation.default.partitions tell Connect to create each topic itself before the first write (the worker’s topic.creation.enable is true by default). Named groups (topic.creation.groups) can give different settings to topics matching a regex. Any topic-level config can be set this way, including cleanup.policy (see the compaction gotcha). Partition count matters twice. It’s the ceiling on consumer parallelism, and every key’s order holds only within its own partition. Choose it before go-live, because adding partitions later remaps keys.

Error handling for source connectors

errors.tolerance (none by default, or all) covers failures in conversion and transformations. Since KIP-779 it also covers producer send failures, and with all, those records are skipped. A source connector has no dead-letter queue: the errors.deadletterqueue.* settings are sink-only. errors.log.enable=true logs each tolerated failure. Failures inside the Debezium task, such as decoding a change, are governed by Debezium’s own event.processing.failure.handling.mode: fail (the default), warn or skip. With none and fail, a bad record stops the task, which for CDC is usually what you want. Skipping a change silently is worse than stopping.

Client overrides

producer.override.* sets the worker producer’s options for one connector. This connector uses producer.override.compression.type=zstd. Overrides are allowed by connector.client.config.override.policy, which is still All in Kafka 4.3. Kafka 4.2 started recommending Allowlist (with connector.client.config.override.allowlist), and the docs say it becomes the default in Kafka 5.0, so list the overrides you rely on now.

The REST API, in the order you’ll use it

Step-by-Step Hands-On

The code is in cdc-with-debezium/ of companion-workspaces. Two Spring Boot 4.1.2 services on Java 25 join the lab: shipment-service writes shipments with Spring Data JDBC, and tracking-projection maintains tracking_view in its own tracking database.

Step 0: The stack

docker compose up -d --wait

Part 1 used only PostgreSQL. The Kafka service runs one KRaft node that is both broker and controller (no ZooKeeper in Kafka 4):

      KAFKA_LISTENERS: INTERNAL://:29092,CONTROLLER://:9093,EXTERNAL://:9092
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka:29092,EXTERNAL://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER

Three listeners: INTERNAL (kafka:29092) for containers on the compose network such as Connect, EXTERNAL (localhost:9092) for the Spring apps on your machine, and CONTROLLER for the KRaft quorum. A client connects to whatever address the broker advertises, which is why containers and the host need different ones. Every internal topic uses replication factor 1 because there is one broker, and KAFKA_AUTO_CREATE_TOPICS_ENABLE is false. The Kafka image is 4.3.2, the latest 4.3 patch; 4.4.0 was published days before this was written, and Debezium 3.7 is built and tested against 4.3.1.

Connect runs the Debezium image (Kafka Connect 4.3.1 inside). GROUP_ID: dispatchly-connect makes the cluster; BOOTSTRAP_SERVERS: kafka:29092 uses the internal listener; and any CONNECT_* variable becomes a worker property, which is how the internal topics get replication factor 1 here. PostgreSQL keeps Part 1’s flags: wal_level=logical, max_replication_slots=10, max_wal_senders=10 (one walsender per streaming connection), and max_slot_wal_keep_size=2GB, the cap that turns a stuck slot into an invalidated slot instead of a full disk. Each service has a healthcheck, and Connect starts only after PostgreSQL and Kafka report healthy.

Step 1: A least-privilege user and a DBA-owned publication

part2/01-debezium-user.sql runs as the database owner. It applies shipment-service’s next migration (two new columns), then sets Debezium up with as little power as possible:

-- shipment-service's next migration: cash-on-delivery amount and free-text delivery notes.
ALTER TABLE shipments ADD COLUMN cod_amount numeric(10, 2);
ALTER TABLE shipments ADD COLUMN delivery_notes text;

-- Consumers order changes by version, so every UPDATE must raise it, including a DBA's hot-fix that
-- forgets to. GREATEST keeps Spring Data JDBC's own increment (it already sets OLD.version + 1).
CREATE FUNCTION shipments_bump_version() RETURNS trigger LANGUAGE plpgsql AS $$
BEGIN
    NEW.version := GREATEST(NEW.version, OLD.version + 1);
    RETURN NEW;
END $$;
CREATE TRIGGER shipments_bump_version BEFORE UPDATE ON shipments
    FOR EACH ROW EXECUTE FUNCTION shipments_bump_version();

-- LOGIN and REPLICATION only: no superuser, no CREATE on the database, no table ownership.
CREATE ROLE debezium WITH LOGIN REPLICATION PASSWORD 'debezium';
GRANT CONNECT ON DATABASE dispatchly TO debezium;
GRANT USAGE ON SCHEMA public TO debezium;
GRANT SELECT ON shipments TO debezium;          -- the initial snapshot reads the table

-- The DBA decides what is published; the connector runs with publication.autocreate.mode=disabled.
CREATE PUBLICATION dbz_dispatchly FOR TABLE shipments;

docker compose exec -T postgres psql -U dispatchly -d dispatchly < part2/01-debezium-user.sql

REPLICATION lets the role open a replication connection and create its slot. That’s all Debezium needs once someone else owns the publication. The alternative, publication.autocreate.mode=filtered, lets the connector create and alter its own publication. It needs CREATE on the database and ownership of every captured table, usually granted through a shared role. That’s convenient when tables are added often; this series keeps publications under the DBA’s control. You’ll see the error the filtered route produces without those grants in the troubleshooting table.

Step 2: The connector

connectors/shipments-cdc.json:

{
  "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
  "tasks.max": "1",
  "plugin.name": "pgoutput",
  "database.hostname": "postgres",
  "database.port": "5432",
  "database.user": "debezium",
  "database.password": "debezium",
  "database.dbname": "dispatchly",
  "topic.prefix": "dispatchly",
  "table.include.list": "public.shipments",
  "slot.name": "dispatchly_shipments",
  "publication.name": "dbz_dispatchly",
  "publication.autocreate.mode": "disabled",
  "snapshot.mode": "initial",
  "decimal.handling.mode": "string",
  "time.precision.mode": "adaptive",
  "heartbeat.interval.ms": "10000",
  "tombstones.on.delete": "true",
  "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",
  "topic.creation.default.replication.factor": "1",
  "topic.creation.default.partitions": "3",
  "producer.override.compression.type": "zstd"
}

Validate it, then register it:

curl -s -X PUT -H 'Content-Type: application/json' \
  localhost:8083/connector-plugins/PostgresConnector/config/validate \
  -d "$(jq '. + {"name":"shipments-cdc"}' connectors/shipments-cdc.json)" | jq '.error_count'
curl -s -o /dev/null -w "HTTP %{http_code}\n" -X PUT -H 'Content-Type: application/json' \
  localhost:8083/connectors/shipments-cdc/config -d @connectors/shipments-cdc.json
curl -s localhost:8083/connectors/shipments-cdc/status | jq -c '[.connector.state, .tasks[].state]'

0
HTTP 201
["RUNNING","RUNNING"]

Connect created dispatchly.public.shipments and __debezium-heartbeat.dispatchly, each with 3 partitions. The four seed shipments arrived as snapshot READ events, spread across partitions by key:

Partition:0  Offset:0  {"id":1}  {"before":null,"after":{"id":1,"tracking_code":"DSP-1001","status":"PICKED_UP",…},…,"op":"r",…}
Partition:0  Offset:1  {"id":3}  {"before":null,"after":{"id":3,"tracking_code":"DSP-1003","status":"PICKED_UP",…},…,"op":"r",…}
Partition:1  Offset:0  {"id":2}  {"before":null,"after":{"id":2,"tracking_code":"DSP-1002","status":"CREATED",…},…,"op":"r",…}
Partition:2  Offset:0  {"id":4}  {"before":null,"after":{"id":4,"tracking_code":"DSP-1004","status":"CREATED",…},…,"op":"r",…}

There’s no order across those four, and there doesn’t need to be.

Step 3: Reading real events

Start shipment-service (./gradlew :shipment-service:bootRun, port 8080) and create a shipment:

curl -s -X POST localhost:8080/shipments -H 'Content-Type: application/json' \
  -d '{"trackingCode":"DSP-2001","hubId":"BLR-07","eta":"2026-10-24T13:30:00Z","codAmount":499.50}'

Part 1 dissected a full envelope; here is the create event’s after and position:

{"before":null,"after":{"id":5,"tracking_code":"DSP-2001","status":"CREATED","hub_id":"BLR-07","eta":"2026-10-24T13:30:00.000000Z","version":1,"updated_at":"2026-10-22T18:29:20.500558Z","cod_amount":"499.50","delivery_notes":null},"source":{…,"txId":776,"lsn":33923968},"op":"c",…}

Status changes go through the endpoint that still holds Part 1’s dual write (save(), then kafkaTemplate.send() to the legacy shipment-events topic, until Part 3). Every API status change in this part looks like this:

curl -s -X PUT localhost:8080/shipments/DSP-2001/status -H 'Content-Type: application/json' \
  -d '{"status":"PICKED_UP","hubId":"BLR-07"}'

The service’s @Version field gave the new row version: 1; every later save adds one. Now the cases that surprise people. Cancel DSP-1002 through the API (DELETE /shipments/DSP-1002), then run part2/02-event-tour.sql, which writes a 9,899-byte delivery note to DSP-1003 and updates its status, updates DSP-1001 under REPLICA IDENTITY FULL, and changes DSP-1004’s primary key from 4 to 104. What came out:

Delete, then tombstone. Key {"id":2} twice: first a d event, then a record with a null value:

{"id":2} | {"before":{"id":2,"tracking_code":"","status":"","hub_id":"","eta":null,"version":0,"updated_at":"1970-01-01T00:00:00.000000Z","cod_amount":null,"delivery_notes":null},"after":null,…,"op":"d",…}
{"id":2} | null

Look closely at before. With REPLICA IDENTITY DEFAULT the WAL only had the key, yet the other fields aren’t null. Those fields are required in the event’s schema, so Debezium fills them with the column’s default (version DEFAULT 0 gives 0) or the type’s zero value ("", and the Unix epoch for updated_at DEFAULT now()). "version":0 here is not the row’s version. Use only the key from a DEFAULT-identity delete.

An untouched TOAST column. DSP-1003’s note is stored out of line (TOAST). The status update that followed didn’t touch it, so the WAL didn’t carry it, and Debezium sends a placeholder:

{"op":"u","after":{"status":"AT_HUB","version":2,"delivery_notes":"__debezium_unavailable_value"}}

A consumer that copies delivery_notes blindly would overwrite real text with that string. Updates made through shipment-service carry the full note: Spring Data JDBC writes every column on save(), so the column counts as touched, and you pay for the full value in every event instead.

REPLICA IDENTITY FULL. The old row comes along, real values this time:

{"op":"u","before":{"status":"PICKED_UP","hub_id":"BLR-07","version":0},"after":{"status":"PICKED_UP","hub_id":"BLR-03","version":1}}

A primary-key change. Three records on the old and new keys, linked by headers (Debezium’s __debezium.context.* headers trimmed):

__debezium.newkey:{"id":104}  {"id":4}    {"before":{"id":4,…},"after":null,…,"op":"d",…}
__debezium.newkey:{"id":104}  {"id":4}    null
__debezium.oldkey:{"id":4}    {"id":104}  {"before":null,"after":{"id":104,"tracking_code":"DSP-1004",…},…,"op":"c",…}

To a consumer, shipment 4 was deleted and shipment 104 was created, possibly on different partitions. The headers are the only link.

Decimals. PUT the same connector with decimal.handling.mode switched to precise (the response is 200, an update), change a status, then PUT the file back:

jq '."decimal.handling.mode" = "precise"' connectors/shipments-cdc.json | curl -s -o /dev/null -w "%{http_code}\n" \
  -X PUT -H 'Content-Type: application/json' localhost:8083/connectors/shipments-cdc/config -d @-

{"status":"AT_HUB","version":3,"cod_amount":"AMMe"}              # precise: unscaled value as base64 bytes
{"status":"OUT_FOR_DELIVERY","version":4,"cod_amount":"499.50"}   # string

"AMMe" is the unscaled integer 49950 as big-endian bytes, base64-encoded; the scale lives only in the schema you switched off. Remember this event: it comes back in step 5.

Step 4: Shaping events with an SMT

connectors/shipments-flat.json is a second, short-lived connector with its own slot (dispatchly_shipments_flat). It shares the publication, which is fine: a publication is just a table list. It uses snapshot.mode=no_data and these transform settings:

  "transforms": "unwrap",
  "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
  "transforms.unwrap.delete.tombstone.handling.mode": "rewrite",
  "transforms.unwrap.add.fields": "op,source.lsn",
  "transforms.unwrap.predicate": "isShipmentTopic",
  "predicates": "isShipmentTopic",
  "predicates.isShipmentTopic.type": "org.apache.kafka.connect.transforms.predicates.TopicNameMatches",
  "predicates.isShipmentTopic.pattern": "flat\\.public\\..*"

• delete.tombstone.handling.mode=rewrite turns a delete into a row with __deleted: "true" and drops the tombstone. The other values are drop, tombstone (the default), rewrite-with-tombstone and delete-to-tombstone. The older drop.tombstones and delete.handling.mode options are gone in 3.7.
• add.fields copies envelope fields in with a __ prefix.
• The predicate keeps the transform off the heartbeat topic.

A status change and a cancellation through the API produced:

{"id":3}  {"id":3,"tracking_code":"DSP-1003","status":"OUT_FOR_DELIVERY",…,"version":3,…,"__op":"u","__source_lsn":34213496}
{"id":1}  {"id":1,"tracking_code":"","status":"","hub_id":"","eta":null,"version":0,…,"__deleted":"true","__op":"d","__source_lsn":34213824}

The heartbeat topic got {"serverName":"flat"} {"ts_ms":1792693980072}, untouched. Remove the demo connector and everything it leaves behind with part2/remove-connector.sh:

part2/remove-connector.sh shipments-flat dispatchly_shipments_flat flat.public.shipments __debezium-heartbeat.flat

{"message":"The Connect framework-managed offsets for this connector have been reset successfully. However, if this connector manages offsets externally, they will need to be manually reset in the system that the connector uses."}
dropped slot dispatchly_shipments_flat
deleted topic flat.public.shipments
deleted topic __debezium-heartbeat.flat

Step 5: The projection

Start it with ./gradlew :tracking-projection:bootRun. tracking-projection reads the raw topic as strings and maps the envelope with a dedicated Jackson 3 mapper (tools.jackson): snake_case names, unknown properties ignored, so a new column upstream doesn’t break it. The listener:

    @KafkaListener(topics = "dispatchly.public.shipments", groupId = "tracking-projection")
    void on(ConsumerRecord<String, String> record) {
        if (record.value() == null) {
            return;   // tombstone after a delete: only there for log compaction; the d event did the work
        }
        var change = changeEventMapper.readValue(record.value(), ShipmentChange.class);
        var applied = switch (change.op()) {
            case "c", "u", "r" -> view.upsert(change.after(), change.source());
            case "d" -> view.markDeleted(change.before().id(), change.source());
            default -> throw new IllegalArgumentException("Unexpected op " + change.op());
        };
        var row = change.after() != null ? change.after() : change.before();
        log.info("{} op={} id={} {} v{} lsn={}", applied ? "applied" : "skipped", change.op(), row.id(),
                row.status(), row.version(), change.source().lsn());
    }

ShipmentChange is a record of records: ShipmentChange(Row before, Row after, Source source, String op, long tsMs). The upsert carries the idempotency rules in its WHERE clause:

                INSERT INTO tracking_view AS v
                    (shipment_id, tracking_code, status, hub_id, eta, version, source_lsn, changed_at)
                VALUES (:id, :code, :status, :hub, :eta, :version, :lsn, :changedAt)
                ON CONFLICT (shipment_id) DO UPDATE
                SET tracking_code = EXCLUDED.tracking_code, status = EXCLUDED.status, hub_id = EXCLUDED.hub_id,
                    eta = EXCLUDED.eta, version = EXCLUDED.version, source_lsn = EXCLUDED.source_lsn,
                    changed_at = EXCLUDED.changed_at
                WHERE v.version < EXCLUDED.version AND NOT v.deleted""")

markDeleted inserts or flags a row with deleted = true and touches nothing else. changed_at is source.ts_ms, the database time, not the connector’s. The projection never reads delivery_notes or cod_amount for its own columns, so the TOAST placeholder can’t hurt it.

The first records it applied:

applied op=r id=1 PICKED_UP v0 lsn=33900608
applied op=r id=3 PICKED_UP v0 lsn=33900608
applied op=r id=2 CREATED v0 lsn=33900608
applied op=r id=4 CREATED v0 lsn=33900608
applied op=c id=5 CREATED v1 lsn=33923968
applied op=u id=5 PICKED_UP v2 lsn=33924584

All four snapshot rows carry the same LSN, the snapshot’s position from Part 1’s hand-off.

Commits. Spring Kafka’s default AckMode.BATCH commits offsets after each poll’s records are processed. A crash mid-batch replays part of it, which the WHERE clause absorbs. Manual acknowledgement buys nothing here: the write is idempotent, so “processed twice” and “processed once” look the same.

Poison records. Boot plugs a CommonErrorHandler bean into the listener container:

    @Bean
    DefaultErrorHandler errorHandler(KafkaOperations<Object, Object> template) {
        var handler = new DefaultErrorHandler(new DeadLetterPublishingRecoverer(template), new FixedBackOff(500L, 2));
        handler.addNotRetryableExceptions(JacksonException.class);
        return handler;
    }

Remember the precise event from step 3? "AMMe" isn’t a BigDecimal, so Jackson threw InvalidFormatException. The handler skipped the retries (JSON that fails once fails forever) and published the record to dispatchly.public.shipments-dlt (Spring Kafka’s default -dlt suffix). The original topic, partition, offset and exception went into its headers. The application logged nothing at ERROR. The view jumped from version 2 to 4: full-row events let the next change repair the state, but the intermediate status was gone, and had that been the last change, the view would have stayed stale. Alert on the dead-letter topic’s growth, not on log lines. Re-driving dead letters once the bug is fixed is safe for the same reason replays are: the version guard ignores anything older than what the view holds.

Step 6: Kill Connect, replay the consumer

Drill 1: a worker dies between offset flushes. Create DSP-2002 and push it through four statuses. Within seconds, before the worker’s next 60-second offset flush, kill Connect without a graceful shutdown, then start it again:

docker compose kill connect
docker compose start connect
part2/events-for-key.sh 6          # every event for shipment 6, in topic order

The first ten lines:

Offset:3  op=c  CREATED  v1  lsn=34214632
Offset:4  op=u  PICKED_UP  v2  lsn=34214944
Offset:5  op=u  AT_HUB  v3  lsn=34215136
Offset:6  op=u  OUT_FOR_DELIVERY  v4  lsn=34215320
Offset:7  op=u  DELIVERED  v5  lsn=34215520
Offset:8  op=c  CREATED  v1  lsn=34214632
Offset:9  op=u  PICKED_UP  v2  lsn=34214944
Offset:10  op=u  AT_HUB  v3  lsn=34215136
Offset:11  op=u  OUT_FOR_DELIVERY  v4  lsn=34215320
Offset:12  op=u  DELIVERED  v5  lsn=34215520

Every event, twice, with identical LSNs. Before the kill, the slot’s confirmed_flush_lsn was 0/20A1040 (34213952): exactly the last offset Connect had stored, not the newest event it had sent. On restart, Debezium logged Received offset commit request on 'LSN{0/20A1040}', but ignoring it. LSN flushing is not allowed yet and Will restart from LSN 'LSN{0/20A12E8}' that is start of the first unprocessed transaction. The slot advances only with Connect’s stored offsets, so the replay window is the offset flush interval. The projection’s log (trimmed to the last first-time event and the duplicates):

applied op=u id=6 DELIVERED v5 lsn=34215520
skipped op=c id=6 CREATED v1 lsn=34214632
skipped op=u id=6 PICKED_UP v2 lsn=34214944
skipped op=u id=6 AT_HUB v3 lsn=34215136
skipped op=u id=6 OUT_FOR_DELIVERY v4 lsn=34215320
skipped op=u id=6 DELIVERED v5 lsn=34215520

Drill 2: a hot-fix that forgets version. On a fresh database, the service took DSP-2001 to versions 1, 2 and 3. Then a hand-written UPDATE shipments SET hub_id = 'BLR-03' WHERE tracking_code = 'DSP-2001', which never mentions version, came back from RETURNING as version 4, and the projection logged applied op=u id=5 AT_HUB v4 lsn=34198776. Without the trigger it would have been skipped. ShipmentPipelineTest keeps this honest (step 7).

Drill 3: replay the whole topic. Stop the projection, rewind its consumer group, and start it again:

docker compose exec -T kafka /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:29092 \
  --group tracking-projection --topic dispatchly.public.shipments --reset-offsets --to-earliest --execute

All 29 records came back. 25 change events logged skipped; the rest were three tombstones and the "AMMe" record, which went straight back to the dead-letter topic. tracking_view was byte-for-byte identical before and after. That’s the property to test for before you go live: replaying the topic is safe.

Step 7: The integration test

ShipmentPipelineTest runs the whole pipeline in containers: postgres:18.6, apache/kafka:4.3.2 and quay.io/debezium/connect:3.7.0.Final on one Docker network. It applies part2/01-debezium-user.sql, registers the same connectors/shipments-cdc.json, and starts the Spring context against them:

    @Container
    static final DebeziumContainer connect = new DebeziumContainer("quay.io/debezium/connect:3.7.0.Final")
            .withNetwork(network)
            .withEnv("BOOTSTRAP_SERVERS", "kafka:19092")
            .dependsOn(kafka, postgres);

Debezium’s own Testcontainers guide still shows Testcontainers 1.x artifacts, JUnit 4 and postgres:15. In 3.7, io.debezium:debezium-testing-testcontainers (incubating) already depends on the Testcontainers 2 modules (testcontainers-kafka, testcontainers-postgresql). Its withKafka(…) now takes a Strimzi test cluster, which is an optional dependency, hence the environment variable above. Three tests check that a status change reaches the view, that a delete is terminal, and that an UPDATE which forgets version still gets through. The run takes about half a minute:

./gradlew :tracking-projection:test

Tips, Tricks & Gotchas

Tip — PUT /connectors/{name}/config, always. It creates or updates in one call, so a deploy pipeline can run the same command forever. POST /connectors fails if the connector exists. Keep the JSON in Git next to the service that owns the table.

Gotcha — event fields you must not trust. A DEFAULT-identity delete’s before holds defaults, not data. Untouched TOAST columns arrive as __debezium_unavailable_value. And a converter or decimal.handling.mode change rewrites every value’s encoding ("499.50" became "AMMe"). Read only the fields you need, and treat connector config like an API contract.

Gotcha — tombstones need compaction to mean anything. Connect created dispatchly.public.shipments with the broker’s default cleanup.policy=delete, so tombstones compact nothing, and after the retention period (7 days by default) the snapshot events are gone: a new consumer can’t rebuild from the topic. topic.creation.default.cleanup.policy=compact (accepted by validate) only applies to topics Connect creates from then on; an existing topic needs kafka-configs.sh --alter --entity-type topics --entity-name dispatchly.public.shipments --add-config cleanup.policy=compact, which the lab ran. Compaction has its own trap: the tombstone, and with it the d event, can be gone after delete.retention.ms (24 hours by default), so a consumer that is offline longer never learns about the delete. Long outages mean a re-snapshot either way (Part 3).

Gotcha — replication factor 1 is a lab setting. With one replica, a broker outage stalls every connector; lose that broker’s disk and connect-offsets (every connector’s position) and connect-configs (the connectors themselves) go with it. Production needs replication factor 3 for Connect’s internal topics and the CDC topics.

Gotcha — minutes of silence after a worker disappears. scheduled.rebalance.max.delay.ms (300000 ms) holds a crashed worker’s tasks in case it returns (Part 3 measured 312 s). Lower it if CDC latency matters more than avoiding needless task moves when a worker merely restarts, and alert on connector lag rather than worker liveness.

Tip — Java 24+ and zstd. Consumers that decompress zstd load native code and warn about restricted methods; add --enable-native-access=ALL-UNNAMED, as the projection’s bootRun and test tasks do.

Gotcha — adding a table to table.include.list doesn’t backfill it. The connector streams the table’s future changes. Existing rows need an incremental snapshot (Part 3).

Debugging / Troubleshooting

Start with the task’s trace (curl -s localhost:8083/connectors/<name>/status | jq -r '.tasks[0].trace' | head -3) and part2/diagnose.sql:

SHOW wal_level;                                            -- must be logical

-- Which tables the connector's publication actually sends (table.include.list can't add to this).
SELECT pubname, schemaname, tablename FROM pg_publication_tables;

-- Who holds each slot, and how far behind it is.
SELECT slot_name, plugin, active, active_pid, wal_status,
       pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn)) AS unconfirmed
FROM pg_replication_slots;

It also lists replica identity per published table and the connector’s sessions: in the lab, two Debezium General client backends and one Debezium Streaming walsender.

Every row below was reproduced in this series’ lab (Part 1 or Part 2):

Key Takeaways

Story Closing

Friday’s demo was two terminals and a stopwatch. Zoya pushed DSP-2002 through four statuses and killed the Connect container. Ten events landed where five belonged, and tracking_view didn’t move. Then she reset the consumer group to zero and replayed everything. Same view, byte for byte.

Tomás was unimpressed in a way that meant he was impressed. “Now break it the boring way.” He stopped the connector and left it stopped. “Your slot’s pinning WAL. Your notifications still come from that send() after save(). And your hubs table was never snapshotted.” He pulled up a calendar invite titled Game day. “Monday. Bring coffee.”

In Part 3, Tomás runs the drills: the transactional outbox that retires the dual write for good, incremental snapshots, slots and heartbeats on a quiet database, schema registries, a two-worker Connect cluster, and the production checklist.

References

1. Debezium PostgreSQL connector (3.7). https://debezium.io/documentation/reference/3.7/connectors/postgresql.html
2. Debezium New Record State Extraction SMT (3.7). https://debezium.io/documentation/reference/3.7/transformations/event-flattening.html
3. Kafka Connect user guide (4.3). https://kafka.apache.org/43/kafka-connect/user-guide/
4. KIP-415: Incremental Cooperative Rebalancing in Kafka Connect. https://cwiki.apache.org/confluence/display/KAFKA/KIP-415%3A+Incremental+Cooperative+Rebalancing+in+Kafka+Connect
5. Spring for Apache Kafka: Dead-Letter Publishing. https://docs.spring.io/spring-kafka/reference/kafka/annotation-error-handling.html

This is Part 2 of a 3-part series: “The Log Never Lies: Change Data Capture with Debezium.”