Debezium change events in Kafka Streams: flatten, route, and survive the delete

You have Debezium streaming a database table into Kafka. Call it shipments: a row per parcel, with an order, a carrier, a weight and a destination, and every insert, update and delete against it landing on a topic as it happens. Now you want to do something with that topic — keep the shipments over ten kilograms, join them to orders, count them per carrier.

If you have not met it, that first sentence is doing a lot of work. Debezium reads a database's own change log and turns every committed row change into a Kafka record, in commit order. No polling, no triggers, no timestamp column that somebody has to remember to touch, and no change to the application doing the writing: the database is already writing that log for its own recovery, and Debezium reads it. Before the live changes it takes a consistent snapshot of what is already there, so the stream starts complete rather than starting from now.

Which log that is depends on the database, and there are thirteen connectors of them: the write-ahead log on Postgres, the binlog on MySQL and MariaDB, the oplog on MongoDB, the redo log on Oracle, and so on through SQL Server, Db2, Cassandra, Spanner and the rest. The documentationis the place to start.

This post uses Postgres. The change events below look almost the same from every connector, so a topology written for one mostly transfers to the next. If you want a real one running locally, docker-compose-postgres.yamlin the tutorial examples takes a few minutes.

Kafka Streams is the natural place for that logic. It is a library that ships with Apache Kafka, not a cluster to run: your application links it, reads the topic, writes another, and the broker you already have does the rest. So you reach for it, and the first record you look at is not a shipment. It is this:

{
  "before": null,
  "after": {
    "shipment_id": "s-1",
    "order_id": "ord-1001",
    "carrier": "DHL",
    "weight_kg": 2.4,
    "destination": "Zagreb",
    "notes": "leave with neighbour",
    "updated_at": "2026-09-19T07:55:40.534185Z"
  },
  "source": { "version": "3.6.3.Final", "connector": "postgresql", "db": "inventory",
              "schema": "public", "table": "shipments", "txId": 826, "lsn": 28126360, "...": "..." },
  "transaction": null,
  "op": "c",
  "ts_ms": 1789804541024,
  "ts_us": 1789804541024594,
  "ts_ns": 1789804541024594256
}

That is the change event envelope: a change is not a row, it is a before, an after and a verb. The new version of the row is in after, the version it replaced is in before, and op says what happened — an insert, an update, a delete. On an insert there is nothing to put in before, so it is null; on a delete there is nothing to put inafter.

Three things follow from that, and everybody does all three. Flatten after. Route by op. Survive the delete. Here is each as one expression, and how to know it is right before it runs against anything that matters.

First, the transform that may end the post here

Debezium ships ExtractNewRecordState, a Kafka Connect single message transform (an SMT, in Connect's words) that unwraps the envelope. Put it on the connector and the envelope is gone: the topic carries the after row, flat, the way you wanted in the first place.

If that is all you need, use it. It runs in the connector, it costs you no application, and this post is over. What it cannot do is anything that depends on more than one record or on more than one topic: routing to different topics by a condition on the data, joining a change stream to another stream, keeping a running count per key, comparing before to after to find what actually changed. The moment the logic outgrows one record at a time, it moves into a stream, and the envelope comes with it.

So: keep the envelope, and learn its three moves.

Flatten after

The expressions in this post are SpEL, written into a node on Alginte's canvas: a mapValues node carries one that shapes the value, a filter node carries one that answers true or false, and value inside them is the record's value, which the JSON converter hands over as a map. Kafka Streams topologies you can draw and runwalks the canvas end to end; the expression reference has the rules. If you write Kafka Streams in Java, read each expression as the one-line body of the lambda in that node.

The column you want is one hop in:

value.get('after').get('carrier')

On the record above that is "DHL". Build the whole flattened row the same way, in a mapValues:

{'id': value.get('after').get('shipment_id'),
 'carrier': value.get('after').get('carrier'),
 'weight': value.get('after').get('weight_kg')}

With the JSON converter the strings are real strings: value.get('op') == 'd'works, no conversion, no toString(). If your connector's value.converter is Avro instead, so the same envelope lands as an Avro record with its schema in a registry, the field arrives as a Utf8: == 'd' still works, .contains('d')does not until you call .toString() on it first.

Route by op

op is one character: c for create, u for update, d for delete, r for a row read during the initial snapshot. Route on it:

value.get('op') == 'c' || value.get('op') == 'u' // upserts
value.get('op') == 'd' // deletes
value.get('op') == 'r' // the snapshot backfill

The one worth a second look is r. When the connector first starts it reads the table and emits every existing row as a change event with op: r, before any live change arrives — with source.snapshot reading first on the first of them and last on the last, if you want to find the boundary. If your logic treats every event as news — send an email, call an API, increment a counter that means "changes today" — the first deployment sends one per existing row. That is the snapshot doing what it says, and it is why the stream starts with history instead of starting empty. Decide what r means to you, and say so in an expression rather than finding out.

Survive the delete

Here is the record your first version dies on:

{
  "before": { "shipment_id": "s-1", "order_id": "ord-1001", "carrier": "DHL",
              "weight_kg": 2.4, "destination": "Zagreb", "notes": "leave with neighbour",
              "updated_at": "2026-09-19T07:55:40.534185Z" },
  "after": null,
  "op": "d",
  "ts_ms": 1789804541026
}

after is null, because after a delete there is no row. Every expression above reads value.get('after').get(...), and on this record that is a method call on null. It fails. Not quietly, not as a false that drops the record: it fails, and depending on where you put it, your topology stops.

The thing to notice is when it fails. It works on every insert and every update and every snapshot row, which is to say it works for as long as nobody deletes anything. Then somebody does.

There are two honest fixes. Route first, and only flatten where you know there is an after:

value.get('op') != 'd' // a filter, upstream of the flatten

Or make the access itself null-safe, with SpEL's safe navigationoperator:

value.get('after')?.get('shipment_id')

On an insert that is "s-1". On the delete it is null, and nothing throws. It reads as a small thing and it is the difference between a topology that survives its first delete and one that does not.

And where the delete is what you actually care about — a downstream store to evict, an index entry to remove — before is where the row is:

value.get('before').get('shipment_id')

Which is why before is worth configuring, and this is the part that surprised me when I measured it. On Postgres, REPLICA IDENTITY decides how much of the old row reaches the connector. With the default, the old tuple is the primary key and nothing else — so an update has no before at all:

{ "before": null, "after": { "shipment_id": "s-2", "carrier": "GLS", "...": "..." }, "op": "u" }

and a delete has a before that looks complete and is not:

"before": {
  "shipment_id": "s-1",
  "order_id": "", "carrier": "", "destination": "",
  "weight_kg": 0.0,
  "notes": null,
  "updated_at": "1970-01-01T00:00:00.000000Z"
}

The key is real. Every other column has been filled in with its type's zero: an empty string for the text columns, 0.0 for the weight, and for the timestamp the epoch, which is the one that catches the eye. Not null, which you would notice — a value, which you would not; only notes is null, and only because that column is nullable. A rule that readsvalue.get('before').get('carrier') gets "" and carries on; a filter onweight_kg < 5 says yes to a parcel that weighed nothing; a rule that readsupdated_at gets 1970.

ALTER TABLE ... REPLICA IDENTITY FULL gives you the whole old row on both updates and deletes, at the cost of more WAL. If anything downstream readsbefore for more than the key, that is the setting, and it belongs in the migration that creates the table rather than in the incident that finds it.

And then the tombstone

There is one more record, and it is not an envelope at all:

key: {"shipment_id": "s-1"}
value: null

After the delete event, Debezium emits a tombstone: same key, null value, no envelope, nothing to reach into. Its job is log compaction — it is the marker that lets Kafka eventually drop every earlier record with that key. It is on by default, and it is the second record your delete produces.

For your topology that means value itself is null. Not after, not a field: the whole value. An expression cannot save you there, because there is nothing to evaluate. Decide what a null value means in your stream and handle it as a null, or turn the tombstones off with tombstones.on.delete=false if the downstream does not need them and you would rather not think about it.

The one that bites silently

Everything above fails loudly, which is the good kind. This one does not.

Give the table a NUMERIC column — a weight, a price, an amount, which in a database is exactly what NUMERIC is for — and under Debezium's default settings it arrives like this:

"weight_kg": "APA="

That is 2.40. Debezium's decimal.handling.mode defaults to precise: a NUMERIC is an exact decimal, a JSON number is a float, and rather than lose precision the connector sends the unscaled bytes, base64-encoded, with the scale in the schema.

What it means for your expression is that weight_kg is a String. So this:

value.get('after').get('weight_kg') < 5

is a String compared to a number. If you want the ordinary thing, say so on the connector:

"decimal.handling.mode": "double"

and weight_kg is 2.4. There is a third setting, string, which gives you"2.40" — exact, readable, and still not a number until you convert it. Choose deliberately: precise when the value is money and the arithmetic is someone else's problem downstream, double when you are comparing weights in a filter and a float is fine.

Prove it before it runs

Every expression in this post is one line, and every one of them has a record that breaks it. The chained get breaks on the delete. The numeric comparison breaks on the default decimal mode. The tombstone breaks anything that assumes a value.

You do not have to find those on a shared cluster. Take a sample of the topic — the ordinary record, an update, a delete, and the tombstone — and evaluate against it while you write. In Alginte that is the playground: one command, its own broker, and a folder where the change events you copied in are produced at every start. There is a Debezium sample in it already, four real events captured from a Postgres, so you can see the envelope without having a database or a connector anywhere near it; the tombstone you produce from the console, with a null value, since a file has no way to hold one.

The part that matters is not the tool, it is the order: write the expression against a record that has actually happened, see the result, and only then deploy. A chained get through a null after is a two-second discovery when the delete is in front of you, and an incident when it is not.

The short version

• Your row is in after; before is the old one; op says which to believe.
• If a flat topic is all you need, ExtractNewRecordState gives you that in the connector and you can stop.
• op: r is the initial snapshot, and it arrives once per existing row.
• after is null on a delete. Route on op first, or use ?., or readbefore — and set REPLICA IDENTITY FULL if you want the whole old row.
• The delete is followed by a tombstone: same key, null value, no envelope.
• A NUMERIC is base64 unless you set decimal.handling.mode.

None of that is hard. All of it is easier to learn from a record in front of you, on a playground you can experiment with freely, before the topology runs anywhere that matters.