DEV Community: alginte software

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

alginte software — Wed, 23 Sep 2026 06:07:29 +0000

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.

A Kafka Streams topology starts in a sandbox, not on your cluster

alginte software — Sat, 19 Sep 2026 17:05:18 +0000

Somebody owns the Kafka cluster you need, and it is not you. You have a topology in mind: read orders, keep the big ones, write them somewhere. You have a predicate in mind too, and you are fairly sure it is right. "Fairly sure" is the problem. To find out, you need to run it, and the only place to run it is the shared cluster: a consumer group in everyone's list, internal topics with your name on them, and an output topic that real consumers may already be reading. Kafka does not take records back. A wrong predicate there means wrong records that somebody downstream has already acted on.

So you do not run it. The idea is shelved, and you never learn that it was the right one.

This post is about the other order: find out first, in a place where being wrong costs nothing, and only then touch the cluster. It needs two things. A stack of your own that you are free to break, and a way to carry what you learned across the border without carrying the stack with it.

The expensive part is not the topology

Writing a Kafka Streams topology is not where the time goes. The time goes into the question the code cannot answer for itself: does the predicate do what you meant, on the records that actually arrive? We made that argument at length in The compiler was never what you wanted: the cost is in the loop between having an idea and finding out, and on a shared cluster that loop is long, so you run it rarely.

The loop is short when three things are true. The records you evaluate against are real. The evaluation you see before deploying is the one that will run after. And being wrong leaves nothing behind. The earlier post was about the first two. This one is about the third, and about what happens after you are right.

A stack of your own, free to break

One docker compose up -d gives you a single-node Kafka, a Schema Registry, and a console that runs Kafka Streams topologies, bound to your machine, with no volumes. It comes seeded with topics, real Avro records against a registered subject, and a topology already running, so the first thing you see is a canvas with numbers on it. docker compose down returns the machine to the state it was in. That is the whole safety story. Nothing here can hurt anyone, because nothing here is shared.

Seeded orders are fine for the first ten minutes. After that you want your own topic. Say it is shipments, with JSON Schema values. Two folders beside the compose file take it in, and because they are read at every up, what you put there survives the down that erases everything else:

playground/
  docker-compose.yml
  schemas/
    shipments-value.json the subject, saved out of your registry
  records/
    shipments.json-schema.jsonl a sample of the topic, one record per line

The file name is the contract. shipments-value.json is registered as the subject shipments-value; .avsc and .proto do the same for Avro and Protobuf. shipments.json-schema.jsonl is produced into the topicshipments, each value framed against that subject, and a line is either a keyed record or a bare one:

s-1|{"shipmentId":"s-1","orderId":"ord-1001","carrier":"DHL","weightKg":2.4,"destination":"Zagreb"}
s-2|{"shipmentId":"s-2","orderId":"ord-1003","carrier":"UPS","weightKg":11.0,"destination":"Milano"}

Keep the topic and subject names identical to the real ones, because the design you are about to make carries them. A small sample is enough, if it is the right one: the ordinary record, and then the ones that break things. A null where a field is optional, an empty list, a value sitting on the boundary of your predicate, a key you did not expect. What you are proving is logic against the shape of your data, and shape is a matter of variety, not volume.

Production may not be able to give you that sample. The ordinary record is there a million times over; the one that breaks your predicate may not have happened yet, and a shared topic is no place to produce it to find out, with everyone else's consumers downstream. Here it is one more line in the file: written by hand, framed against the same subject as the sampled ones, seen by nobody's consumer but yours.

Anything personal in the sampled records is yours to redact before it lands in the folder. The playground's repository has these two files as templates to copy.

Now the loop. Drop a source node on shipments, bound to shipments-value. Add a filter, and start typing the predicate. As you type, it is evaluated against the newest record on the topic, and the result sits next to the editor: the input record, the output, the verdict. Put a mapValues in front of it and the filter sees the shape themapValues produced, not a guess. Misspell a field and the editor tells you on the keystroke. On an Avro or Protobuf record the access fails, with the same message the deployed operator would throw, because it is the same evaluation. On a JSON Schema record, where a missing key is a quietnull at runtime, the editor warns on the name and offers the closest one the subject declares. This is the short loop: the few seconds between the typo and finding it.

Fix the name, finish the predicate, and the same panel judges it against a record you brought:

Submit, and a gate runs before anything deploys: a source topic that does not exist, a source whose subject the registry does not have, join inputs that are not co-partitioned, a windowed key on its way to a sink that cannot write it. Each refusal names the node and the fix. Pass the gate, and the stream runs here, on this broker. Produce a few records, written by hand or generated at random from the subject's schema, one click each. Watch the sink fill, open the State tab and see the threads, the stores and the lag. Delete it, keep the state or remove it. Deploy it again under the same id, and the console tells you what the earlier run left on the cluster, a changelog and a committed offset, and asks whether to continue on it or start clean.

None of this is a simulation. It is the Kafka Streams client, running your topology, on a broker you can throw away. The console in the loop is Alginte, in the mode it callsthe playground, and the one claim that makes the loop worth trusting is the one we test hardest: an expression that evaluates one way in the editor evaluates the same way deployed, on Avro, JSON Schema and Protobuf alike. That promise is held by a corpus of rows that run each expression on both surfaces and refuse to ship a disagreement.

Carrying it across the border

A design that works here has to reach the cluster that matters, and it must not bring the sandbox with it.

The design travels as a file. Download writes the whole thing: the configuration, every node with its expressions and serdes, the edges, the layout. The file is named after the application.id, and it belongs in git, beside the code it feeds. It is plain, indented JSON, so a changed expression is a one-line diff, a new node is a block a reviewer can read in a pull request, and the history of the stream is the history of the file:

   "data": {
     "name": "light DHL parcels",
     "expression": {
- "expression": "value.get('carrier') == 'DHL' && value.get('weightKg') < 5",
+ "expression": "value.get('carrier') == 'DHL' && value.get('weightKg') < 10",
       "language": "SpEL"
     }

That is what makes it the durable artefact: the sandbox is thrown away, the file is not.

What the file does not carry is as important: not the records, not the topic, not the schema, not the stream's state. Those are the cluster's, on both sides.

On the console that runs against the real cluster, Upload the file. The same wizard opens with the same validation. Three things are different at the border, and each is handled by a rule rather than by remembering.

A design carries its logic, not its cluster. If the file hasbootstrap.servers, a registry URL or security settings in it, the upload leaves them out and says which, so the connection that applies is this console's, never the sandbox's.

Names are checked against this cluster. The gate refuses a source topic that does not exist here, and a source whose subject this registry lacks, naming the node. Register the subject first, with the same text you put inschemas/, and the refusal goes away.

The id is checked against this cluster's memory. If a stream under thisapplication.id has run here before, the console says so before deploying: these internal topics, this committed offset. Continue on them, or start fresh. It never guesses.

Then Submit. The deploy is the same gate and the same client. The only thing that changed is the cluster it joined.

That is one of two routes out of the sandbox, and neither is the lesser one. The other: your team ships its streams as Java services, with a compiler and a test suite, and what you carry across is the shape, the expressions and the evidence that they are right. Either way you arrive having already answered the question the compiler cannot.

What the sandbox does not prove

Scale. A design proven on a small, well-chosen sample is proven for what it does to a record, not for its throughput, its partition layout under load, or how long its stores take to restore after a restart. Those are questions for the real cluster, and there they have answers. In the sandbox, the State tab showed a handful of records moving. On the real cluster, the same tab shows what matters at scale: a restore as it replays, the threads and the tasks they hold, the lag of every store, and on the canvas the throughput of every node. It is the same console on both sides of the border: the workbench here, the instrument there. The sandbox tells you the logic is right. The first hour on the real cluster tells you the rest, and it is an hour you can watch rather than infer.

The order of operations

Start where being wrong is free. Bring the shape of your data, and a sample that covers its corners. Write the expression against a real record and let the editor tell you on the keystroke. Deploy on a broker you can throw away, and throw it away. Then, if you choose to run it as it is, carry the design across to the production cluster as a file, and let the gate on the other side check the names, the id and the connection. Deploy there once, on purpose.

The production cluster is still somebody else's. The difference is that your stream now arrives there with its logic already proven, and "fairly sure" left behind in the sandbox.

Monitoring Kafka Streams applications: the state a cluster view cannot show

alginte software — Tue, 15 Sep 2026 05:38:25 +0000

A typical Kafka console reads the cluster: topics, partitions, consumer groups, committed offsets, lag. That is the right picture for most of what a console is for.

A Kafka Streams application is a client of that cluster, and from the cluster's side it looks like what it is: a consumer group with members, some committed offsets, a few internal topics with a naming pattern.

What the application is doing is not on the cluster. Whether it is restoring a store from its changelog, and how far along it is. Whether a standby is keeping up with its active. Whether the threads Kafka reports as alive are serving the tasks assigned to them. How many times it has rebalanced since it started and how long the last one took. All of that is inside the process, in listeners and gauges the Kafka Streams API exposes to the application itself.

The restore case is the sharpest. The consumer that replays a changelog is not a member of any group, commits no offset, and keeps its position in a local checkpoint file. So while an instance that started with an empty state directory replays millions of records before it can process a single new one, the cluster shows a stable group and, if input keeps arriving, a lag that grows for no reason it can show. The application is doing the one thing that decides when it will be available again, and the cluster has no way to say so.

Anyone who has explained that silence to a colleague, or to a customer on an upgrade call, knows the shape of the problem: the application was fine, or was not, and the picture in front of everyone was of something else.

We built Alginte around the stream because of this gap, on both sides of it.

Visibility into the stream's shape

The first side was design time. A Kafka Streams topology is a graph, and the DSL, the describe output and every diagram in every article draw it as one, but the drawing never contains what the operators do. So we build the stream as the graph it is: in the modeler you draw the topology, node by node and edge by edge, and the boxes get their bodies.

Each operator carries the expression it applies, and the editor shows a record that is really in the source topic and what that expression makes of it. The derived shape at every node is marked as inferred rather than asserted. You see the stream's shape, and what it does to your data, before it exists on the cluster. That is visibility into the stream as a thing you are building.

Visibility into the stream's operation

The second side is run time, available since 0.11.0. Because the console runs the stream, it holds the Kafka Streams client and can read it directly: no agent, no scraping, nothing written to the cluster.

A State tab on every deployed stream shows what the process knows. A restore banner appears only while a store replays its changelog, showing how many of the records it has replayed.

The lifecycle is the state the console derives, not only the one Kafka reports, because the raw client state misleads in two known ways: the console reads REBALANCING with every thread dead as ERROR, and RUNNING with assigned tasks served by nobody as DEGRADED, with the shortfall in red.

Every store partition is listed with the one figure that matters most, read by its role: records left to replay while restoring, distance behind the active on a standby, and the plain changelog lag at steady state. Threads, rebalances with their durations and an event timeline complete it. Beside all of it sit the cluster's own facts about the same stream, group state, committed offsets, changelog topics, so the two views can be read against each other.

A stream replaying 7.5 million changelog records after its state directory was wiped. The cluster's view of the same moment is a stable group and no sign of why.

The same page opens for a Kafka Streams application the console does not run, on any cluster it reads with the usual read-only rights. There it shows what the cluster can say, the group, the offsets and lag, the changelog topics and how far each reaches, and marks the rest as unseen: no lifecycle, because a group reads stable while threads die and we would rather show absence than invent one; no restore progress, no store sizes. Every figure the observer cannot see is a dash, never a zero. The docs describe which view you get in each case.

An application this console does not run: the cluster's facts, and everything it cannot see marked as absent rather than zero.

That is the depth the cluster allows, and it is where the idea points next. An application you run outside Alginte will be able to report its own state to the console, thread state, restore progress and store sizes included, through a small reporter library added to the application, one call before it starts. The console side of that, an endpoint that accepts what a runtime pushes, exists today; the library follows. And the streams the console runs need not run in the console's own process forever: a headless runtime that hosts them elsewhere and reports back to the same page is the shape we are building toward. Either way the page stays the same. What changes is how much of the stream it is allowed to see.

Together the two sides are one idea. The modeler lets you see the stream you are building; the State tab lets you see the stream you are running. Same object, both ends of its life, with the console built around it. Martin Kleppmann turned the database inside out; this is the console turned inside out, the stream in the middle and the console around it.

What seeing it does not change

Seeing a restore does not make it faster. A large store still takes the minutes it takes to replay; what changes is that those minutes are a number on a page with a bar beside it instead of an unexplained silence. Seeing a dead thread does not bring it back; what changes is that the person who has to decide what to do can tell it from an ordinary rebalance.

See what it is doing

A stateful stream application is a commitment: it holds state, it has to restore it, and it can fail in ways the cluster does not show. Run it where you can see it, so that its state is a page anyone on the team can read, not a question for whoever wrote it. That is what the console is for: a stream you can see while you design it, and a stream you can see while it runs.

The compiler was never what you wanted

alginte software — Fri, 11 Sep 2026 06:55:08 +0000

You have an orders topic on a Kafka cluster, its values encoded with Avro against a schema in the Schema Registry. You want the orders worth more than fifty euros on a topic of their own, and you have decided to do it with Kafka Streams — a JVM library, your code, your deployment.

The schema has five fields:

{"type": "record", "name": "Order", "namespace": "com.alginte.demo",
 "fields": [
   {"name": "orderId",    "type": "string"},
   {"name": "customerId", "type": "string"},
   {"name": "item",       "type": "string"},
   {"name": "quantity",   "type": "int"},
   {"name": "priceEur",   "type": "double"}]}

You want one line of logic over them: quantity * priceEur > 50.

Here is everything standing between that line and a topic of big orders.

Seven steps

The route Confluent's own examples take, and many projects with them:

Get the schema out of the registry and into your repository as an .avsc — or, if your team owns the schema in the repository and publishes it to the registry, the other way round. Whichever copy you call the source, there are now two that can disagree.
Add the code generator to your build.
Configure it — source and output directories, and the string type.
Build, producing Order.java under target/generated-sources.
Write the topology against the generated class.
Package the application, with the schema, the class and the serde.
Deploy it somewhere that runs a JVM.

Steps 2 and 3 are this, once — in Maven, though Gradle's equivalent has the same shape:

<plugin>
  <groupId>org.apache.avro</groupId>
  <artifactId>avro-maven-plugin</artifactId>
  <version>1.12.1</version>
  <executions><execution>
    <phase>generate-sources</phase>
    <goals><goal>schema</goal></goals>
    <configuration>
      <sourceDirectory>${project.basedir}/src/main/avro</sourceDirectory>
      <!-- without this, string fields generate as CharSequence, not String;
           Confluent's own examples set it for the same reason -->
      <stringType>String</stringType>
    </configuration>
  </execution></executions>
</plugin>

And step 5 is the part you actually wanted to write:

builder.stream("orders", Consumed.with(Serdes.Void(), orderSerde))
       .filter((key, order) -> order.getQuantity() * order.getPriceEur() > 50)
       .to("big-orders", Produced.with(Serdes.Void(), orderSerde));

Seven steps, one of them the predicate. Two of them — adding the plugin and configuring it — you do that once.

The filter is deliberately trivial — a real topology joins, aggregates and branches — but the seven steps are identical for twenty operators, because they are charged per project rather than per line of logic. And application.id, bootstrap.servers, the registry URL and your cluster's authentication are missing from the list because they are the price of running against a real cluster and registry, not the price of generating classes.

Those two are not the cost.

Then you want to use a new field

Somebody adds region to the schema. While it sits there unused you are fine — Avro resolves the writer's schema against yours, your generated class does not know the field exists, and nothing needs rebuilding. Schema evolution is doing its job.

Then somebody asks for EU orders only.

The change to your logic is one term: && order.getRegion().equals("EU"). The change to your project is steps 1, 4, 5, 6 and 7 — pull the new schema, regenerate, rewrite, repackage, redeploy.

That is the actual price, and it is charged not per schema change but per schema change you need. Which, over the life of a pipeline, is many of them: fields get added because somebody intends to use them.

What the seven steps buy

They buy the compiler. order.getQuantiy() does not compile. Rename a field, regenerate, and every stale use site turns red before anything runs. The IDE completes field names. Refactoring works.

That is worth having. It is also answering a narrower question than the one you actually have.

The compiler can tell you that getQuantiy() is not a method. It cannot tell you whether quantity * priceEur > 50 is the predicate you meant, whether it matches any record on the topic, or whether the field you are multiplying holds what you think it holds. For that, the seven steps have one answer: deploy it and look.

So the loop you are really in is not edit, compile. It is edit, compile, package, deploy, produce a record, read the output — and it costs the same whether the expression was right or wrong.

A tighter loop

There is another shape for this, and it is the one we build: Alginte, a browser-based topology builder that assembles it at runtime from topics you pick, instead of compiling it into an application you ship.

Point it at orders. The schema comes from the registry at runtime, the five field names arrive as completions, and the predicate is a string — written in SpEL, Spring's expression language, evaluated once per record:

value.get('quantity') * value.get('priceEur') > 50

That expression is evaluated against a real record from the topic while it is being typed — a record off the partition rather than a mock or a fixture, with the answer beside it.

When it is right, it deploys as a Kafka Streams topology: the same library, the same KafkaStreams client, the same rebalances, state stores and changelog topics you would have got from the seven steps. No separate engine is involved. The only thing that changed is how the topology was written.

No .avsc in a repository, no plugin, no target/generated-sources.

The record on the in line is a real one off orders — quantity 3, priceEur 17. The out line is whatever the expression currently returns, and it follows the expression as it is edited: 51, which is the number the > 50 filter is about to judge. Nothing has been deployed.

Mistakes surface in the same place. A method that does not exist on the type the record actually carries is reported while it is being typed, naming the type — the post about building that editor has that one on camera.

That is not a compiler. It is the question the compiler could not answer, asked against real data, answered in seconds. And when somebody adds a sixth field and you want to use it, you type its name.

What you give up

The compiler.

value.get('quantiy') is a valid expression — javac never sees it, so nothing rejects it before it runs. Rename across the project goes too, and the unit tests that constructed Order objects.

What you do not lose is the typo itself. On Avro a misspelt field throws at access — Not a valid schema field: quantiy — so it fails in the editor and fails deployed, identically. That is not the compiler's guarantee, which is made before anything runs. It is the same failure arriving in both places at the same moment: a weaker promise, and a real one.

Some of it returns in a different form — field names completed from the registry, so the typo is never offered; the expression checked against a real record as you type; the same failure surfacing in the editor that would surface in the deployed topology. A different guarantee, weaker in some places, stronger in one.

What does not go is the review. A topology drawn here exports as JSON and imports back, so the thing a reviewer reads and the thing git keeps is a file, not a browser session — the build step is gone, the artifact is not.

You can have the generic half of this without any of the rest. Nothing stops you writing a Java topology that reads GenericRecord and never generates a class: steps 1 to 4 disappear, and so does the compiler, since record.get("quantiy") is a string lookup that javac will not check either. What you are left with is the worst of both — no type safety, and steps 6 and 7 still in front of you.

The generic types are not the point. The loop around them is: the completions, the record on screen, and the deploy that is a click rather than a pipeline.

What that buys is a faster answer to the question you actually had.

And if you would rather write the Java regardless — the stream is one part of a larger application, or your team works that way, or any of the other good reasons — the loop is still worth having first. Draw it here, get the expressions right against real records, then spend the seven steps on logic you already know works. That is a better use of an afternoon than finding out after the deploy.

This came out of building Alginte, a visual Kafka Streams builder — self-hosted, free to run.

SpEL in Kafka Streams: from evaluator to editor

alginte software — Wed, 09 Sep 2026 09:42:51 +0000

A Kafka Streams application is two things: a graph, and the logic inside it.
The DSL writes both in one chain — operators like mapValues, filter and
join become the nodes, the chaining becomes the edges, and the lambda inside
each operator is where the work happens:

record Enriched(String customer, String product, double total) { }

builder.stream("orders", Consumed.with(Serdes.Void(), orderSerde))
       .mapValues(value -> new Enriched(
               value.getCustomerId(),
               value.getItem(),
               value.getQuantity() * value.getPriceEur()))
       .filter((key, value) -> value.total() > 50)
       .to("orders-enriched", Produced.with(Serdes.Void(), enrichedSerde));

A graph is a picture, so we draw it: operators dragged onto a canvas and wired
together.

The same four operators as the Java above, drawn rather than chained — one box
each for stream, mapValues, filter and to. The two that need an
expression each carry a red badge, and the step above counts them; the source and
sink already have their topics. The graph is finished and the contents are not.
Watching it built, and what happens after, is the subject of an earlier post.

The body of an operator is not a picture. It is a small computation over one
record — and it is the only part of the whole thing doing anything specific to
your data. That part, and only that part, is what an expression language is
for.

Written as an expression, that body needs no class and no build around it:

{'customer': value.get('customerId'),
 'product': value.get('item'),
 'total': value.get('quantity') * value.get('priceEur')}

The filter's body is one line: value.get('total') > 50.

Writing those bodies in Java is not the hard part. Finding out whether they are
right is the hard part: compile, package, point the app at a broker, produce
a few records, read what came out. Every step is routine, the whole loop is
slow, and it takes exactly as long whether the expression was correct or not.

Evaluated as an expression against a real record while you type, that loop
collapses. It widens the set of people who can experiment with stream
processing, and it makes learning it cheap — the barrier was never the DSL, it
was everything you had to stand up before the DSL would run.

Why SpEL

It is built for this.
SpEL has driven routers and transformers in Spring Integration for years: a language built to take expressions at runtime.

It is already there. In a Spring application it arrives on the classpath with the framework.

It can be contained. A Kafka Streams lambda is compiled by whoever owns the deployment — your classpath, your risk. An expression typed into a browser arrives at runtime and runs in a process the author may not own. So containment comes first: a restricted evaluation context, with no type references, no constructors and no bean resolution. Spring provides that deliberately, which is what makes the whole approach possible — though the honest word is containment rather than safety.

There is a tokenizer to reuse. That expression needs a real editor around
it, and what that editor shows has to agree with what will actually run. The
surest way is to tokenize with the same code the evaluator uses. SpEL has one.
Getting hold of it is where this gets interesting.

Where the knowledge comes from

In the Java above, the knowledge came from types — a class on the classpath, there at build time, whether generated from a schema or written by hand.

An expression editor in a browser has none of it. No compile step, no class to consult — the only things that know anything about the data are the Schema Registry and the records sitting on the topic. Everything the editor offers has to come from those two, while someone is typing.

Drawing the graph is the easy half, and on its own it buys very little: a
canvas with a bare text box for the logic trades away the help the IDE was giving and offers nothing in its place. Separating the two only pays off if the expression half gets something of its own.

It will not be an IDE — but it has what an IDE cannot: the data. An IDE knows priceEur is a number; it does not know this record's is 17. One expression, evaluated against a real record as you type, is closer to a REPL than to autocompletion.

The editor is Monaco, the
component behind VS Code, so it already knows how to highlight tokens, show a completion list and put a marker on a line. What it knows nothing about is SpEL, or your data.

Completions from types, and from content

Completions come from two different places, and the difference is worth keeping visible.

At a source node — reading a topic whose values have a registered schema — the completion list is the schema's own fields with their declared types. That is a contract: something else asserted it, and it holds for every record on the topic.

One operator downstream, after a mapValues, there is no schema any more. The value is whatever the expression returned. The fields offered there are derived by evaluating that expression against a real record and looking at what came out — and the UI marks them (inferred), because that is what they are: an observation of the sample in hand, not a promise. Draw another record and they may differ; if the data is heterogeneous, no single sample will say so.

Inference is not the last word. A node can declare its output in the same
vocabulary a source uses — a serde per side, optionally a registered subject — and a node that declares one becomes a fresh start: its completions are a contract again. What you cannot do is type a list of field names, and that is deliberate. Fields belong to schemas, and a shape worth describing is usually a shape worth registering.

The same dropdown at two different nodes, two kinds of knowledge, labelled differently. It would be easy to present both as "fields", and it would be wrong.

An empty topic is not a dead end. With no record to draw, the sample is
randomly generated from the registered schema instead, and the preview says which one it is holding. That keeps the editor working before any data exists — while being the weaker evidence of the two, since it demonstrates the shape rather than your data.

One operator after the mapValues, the filter offers customer, product and total — none of which exist in any schema. They are there because the expression on the node behind was evaluated against a real record, and the type hint says so. In the source node the same list would have come from the registry instead. The earlier post has this as a
clip.

What an expression receives, and returns

All of that depends on what the deserializer actually hands over — and that is rarely the Java type you would guess from the schema. Avro gives you a GenericRecord, and a field the schema calls a string does not arrive as a String: it arrives as org.apache.avro.util.Utf8, a CharSequence wrapping the raw bytes. Equality still works — value.get('item') == 'Kettle' matches on the orders stream — because SpEL compares CharSequence content. .contains(...) does not, because it is a String method that CharSequence never declared, so it needs a .toString() first. That is the kind of detail that lives in serde Javadoc rather than anywhere you would think to look, and
it costs an afternoon the first time.

Typing value.get('item').contains('Kettle') into a map expression; the editor reports one issue; opening the status shows EL1004E, method contains(java.lang.String) cannot be found on type org.apache.avro.util.Utf8; adding .toString() clears it to no issues.

The same afternoon, spent in under thirty seconds. .contains(...) on an Avro string, the message naming org.apache.avro.util.Utf8 rather than "invalid expression", and .toString() clearing it.

Types are one half of what an editor can offer; the record is the other. Each node's sample preview puts that record in and the evaluated output out, side by side, re-running as the expression changes — so what an expression actually returns is on screen while it is still being written, against the data it will run on.

The Sample preview panel on a map node, showing a real record drawn from the orders topic on the in line. Typing value.get('item') makes the out line read Kettle; replacing the expression with value.get('quantity') * value.get('priceEur') makes the out line read 51. The editor reports no issues throughout and nothing is deployed. in is a real record off orders; out is whatever the expression returns, and it follows the expression as that changes — value.get('item') gives "Kettle", then the intro's own quantity * priceEur gives 51, the value the
> 50 filter downstream exists to judge. Nothing has been deployed, and the editor never claims either answer is correct. It only makes them impossible to miss.

Avro's type names leak into error messages too, which the next section has to deal with.

Errors a user can act on

SpelCompilerMode.IMMEDIATE exists to make expressions faster by compiling them to bytecode. We use it in the validator, as a type checker. Compiling forces type resolution to happen while someone is still typing, which is exactly when you want to hear that a method does not exist on the type you actually have.

Catching it early, at the right line and column, is half of a diagnostic. The other half is what it says.

SpEL's messages assume a developer who knows the JVM types involved. Ours are read by someone who has never heard of GenericData$Record and should not have to. So the validator rewrites them.

An indexing failure — someone reasonably tries value['item'] — becomes:

indexing with ['...'] is not supported on schema-backed records — use
get('fieldName') instead (the completions suggest it). A deployed topology
fails the same way on Avro / Protobuf values.

The fix, a pointer at where it is offered, and a promise about the
runtime. That last clause matters most: it tells the reader this is not an editor quirk to be worked around, because the deployed topology fails
identically.

Everything else gets a blunter rule: any message mentioning an internal sample type has that type replaced with "the sample record". Someone who writes a bad expression should learn what they did, not what class the deserializer happened to return. A few cases are worth naming precisely and get hand-written messages, like "Invalid map access syntax. Use value['key'] without a dot."

The rewriting covers Avro's generic types, Jackson's nodes and — since 0.10.0
— Protobuf's message types, which became reachable the moment get('field')
started evaluating on them.

The tokenizer you cannot reach

An editor needs three things an evaluator does not provide: syntax
highlighting, completions, and errors at the right line and column. All three start from tokens.

Spring has a tokenizer — the one its own parser uses. It is package-private.
Against Spring Framework 7.0.8:

class org.springframework.expression.spel.standard.Tokenizer {
  public org.springframework.expression.spel.standard.Tokenizer(java.lang.String);
  public java.util.List<org.springframework.expression.spel.standard.Token> process();
  static {};
}

The constructor and process() are already public. The class is not, so
from outside the package there is no way to call them.

The public surface of that package is SpelExpressionParser, SpelExpression
and SpelCompiler. Parsing, evaluation, compilation — no tokenization.

SpelExpression does expose a parse tree — getAST() returns SpelNodes with
start and end positions, which looks like exactly the supported route we missed.
But a parser needs a valid expression, and an editor spends most of its life
holding an invalid one. value. is precisely when completions have to fire, and
it does not parse; nor does this post's own map literal before its closing
brace. Both tokenize.

So we copied Tokenizer, Token and TokenKind from
Spring's spel.standard package,
under Apache-2.0, into a package of the same name in our own tree — which is
what puts those package-private types within reach.

All of it exists to enable one line:

List<Token> tokens = new Tokenizer(inputData).process();

What we add on top is thin, and most of it is a coordinate transform. Spring
reports absolute character offsets; editors want line and column. So each
Token becomes a SpELToken(start, end, line, charPositionInLine, type, modifiers) — the shape Monaco and LSP want — and goes over a WebSocket to the
browser. There they are served through a DocumentSemanticTokensProvider,
which is the highlighting: every colour in the expression comes from the same
tokenizer the evaluator parses with, so what the editor shows and what runs
cannot drift apart.

That is the whole of it. Spring's authors had already written the hard part;
nothing was missing but a way to call it. What we carry is not
complexity — it is their file.

What it costs

SpEL is not Java. Broadening who can write a transformation costs the type
system, the IDE, refactoring, unit tests, and the entire library ecosystem, in
exchange for a string in a text box. That trade is defensible for "what does
this do to my records" and indefensible for a thousand-line pipeline.

The editor cannot catch a wrong answer. It catches expressions that are
invalid — a method that does not exist, a syntax it cannot parse. One that is
type-correct and simply always false passes every check. The preview above will
faithfully show you the wrong answer; nothing will tell you that is what it
is.

Package-private internals carry no compatibility promise. Our copy of the
tokenizer can break between Spring versions with no deprecation cycle, and that
is our problem rather than Spring's. We took the risk knowingly; it is still a
risk.

Protobuf was opaque when this was first published — a DynamicMessage
offers no field access by name — and unlike the rest of this, it was ours to
fix. It shipped in 0.10.0: one SpEL MethodResolver gives
get('field') the same meaning on Protobuf as on Avro and JSON Schema,
registered at all three places an evaluation context is built, so the editor
and the runtime cannot disagree about it.

Is there a better way to do this?

That is our workaround, and we are not claiming it is the right answer.
Package-private is a deliberate choice — internals stay private precisely so
they can change — and we went around it.

So the question is genuine. If there is a supported path to a SpEL token stream
that we missed, we would much rather be told than go on carrying a copy of
somebody else's tokenizer. And if there is not, we are probably not the last
people who will want one.

Either way the pieces are all there: the tokenizer works, the errors carry
positions, the compiler doubles as a type checker. We would just like a
supported way to hold one of them.

Updates

2026-09-07 — when first published (2026-08-31), this post said Protobuf values were opaque to the editor and that the fix was on the list. It shipped in 0.10.0 on 2026-09-03: get('field') evaluates on Protobuf in the editor and deployed, and Protobuf's type names joined the message rewriting. The two passages above were corrected to say so.

This came out of building Alginte, a visual Kafka
Streams builder — self-hosted, free to run.

Kafka Streams topologies you can draw and run

alginte software — Tue, 08 Sep 2026 13:33:25 +0000

You have an orders topic. Each record carries a customer, an item, a
quantity, and a price. You want the big ones — total over 50 — in a topic of
their own, with that total already calculated.

In Kafka Streams, that is a mapValues and a filter: a dozen lines of real
logic.

Getting them to run is a different size of job: a build file, a serde
configuration, a jar, somewhere to put it, and a redeploy every time you want
to check if your expression was right about the data. The work is small. The
apparatus around it is not — and none of it tells you anything until the whole
thing is up.

That distance is the subject here. Not that Kafka Streams is hard, because it
is not — but that the trip from knowing what you want to watching it
happen is longer than a dozen lines of logic deserves.

Start with what Kafka Streams already gives you — most of the picture is
there, and the missing part is smaller than it looks. Deploy a topology and
ask it to describe itself; here is a real one: four operators, reading
orders and writing orders-enriched:

Topologies:
   Sub-topology: 0
    Source: Orders-src (topics: [orders])
      --> order-total-enrich
    Processor: order-total-enrich (stores: [])
      --> big-orders-bigOnly
      <-- Orders-src
    Processor: big-orders-bigOnly (stores: [])
      --> Big-Orders-sink
      <-- order-total-enrich
    Sink: Big-Orders-sink (topic: orders-enriched)
      <-- big-orders-bigOnly

That is a graph. Nodes, directed edges, a topological order. The DSL
constructs one; describe() prints one, and it is what anyone sketches on a
whiteboard when explaining a pipeline to someone else.

Look at what it describes. order-total-enrich is a processor: its name, its
position, what feeds it, and what it feeds. What it does lives somewhere
else — the lambda inside mapValues, the entire reason that node exists, is in
your source rather than in the description. A topology description captures
shape, and captures it completely; behaviour was never its job.

What falls between them is the interesting part. A visual representation of a
Kafka Streams application — ours included, until recently — tends to be a
picture of the shape. Fill the boxes with real bodies and the picture stops
being documentation of the program and becomes the program — something you can
draw and run.

The data is already there, and so is its schema

When records are produced through a schema registry — as these were — each one
carries the id of the schema it was written against. So the shape is described
before any topology exists, and a record and its schema are connected before
anyone draws anything — you can watch it work.

Reading the orders topic with a Void key deserializer and Avro value deserializer, expanding one record to its decoded JSON, then following the Schema tab through to the orders-value subject.

Note what the value deserializer is not asked for: a subject. The decode
succeeding is the connection; the schema tab afterwards only names it.

(Void on the key side because these records genuinely have null keys —
anything else renders noise where there is nothing.)

What goes in the boxes

So what should order-total-enrich produce? A customer, a product, and that
total. In a Java project, you would write that as a lambda and compile it; here
it is an expression, and this is the whole of it:

{'customer': value.get('customerId'),
 'product':  value.get('item'),
 'total':    value.get('quantity') * value.get('priceEur')}

Evaluated once per record, inside the same JVM that is running the topology.
The parse happens up front, and the parsed expression is cached, so the
per-record cost is evaluation, not parsing.

Reaching for an expression language rather than something more powerful is not
a simplification — it follows from who writes it. A Kafka Streams lambda is
compiled by whoever owns the deployment, so trust is implicit and never has to
be examined. Move authoring into a console and the author is a user: the code
arrives at runtime, from outside, and the process it runs in is yours. The
obvious alternative — letting people upload compiled code — answers the same
need by handing a stranger arbitrary execution inside that process.

We use Spring's SpEL
for this. It is embeddable, it has an evaluation context you can restrict
deliberately rather than by accident, and in a Spring application it is already
on the classpath.

Which makes the sandbox a precondition rather than a feature. Expressions
evaluate against a restricted context: no type references, no constructors, no
bean resolution. The honest word for that is containment — we can tell you
exactly what is blocked, and we cannot prove that nobody will find a way
past it.

Drawing it, and watching where the type survives

Now the graph. Four operators dragged out, wired, named, and pointed at
orders:

Building the topology on a canvas: dragging source, mapValues, filter and sink nodes, wiring them, naming them, and binding the source to the orders topic.

The beat worth pausing on is the source's serde section, about a third of the
way in. The value side fills itself in — Avro, subject orders-value —
because a registered subject makes the value type a contract. The key side
stays empty until you say so, because there is no orders-key subject and nothing
to derive it from.

From the source down, each operator either preserves the value or replaces it.
filter, peek, repartition, toStream pass it along unchanged; mapValues
ends the contract, because after it the value is whatever your expression
returned. That is not a property of any tool — it is how the DSL works, and it
is the rule anything deriving types has to follow.

The loop

With a contract at the source and a rule for how it propagates, the editor can
say something useful while you type — and a real record underneath answers
back. Type, look, adjust: that is the loop.

Writing a mapValues expression with completions drawn from the registry schema, watching a real record flow through the preview, then a filter whose completions offer the computed total field.

Two different kinds of knowledge appear in that clip, and the difference
matters.

In the mapValues node, the completions are the schema's own fields —
customerId, item, quantity, priceEur, with their declared types. That
is a contract; it came from the registry.

One node downstream, the filter offers total. There is no total in any
schema, in any topic, anywhere. It exists because the expression above computed
it, and the shape was observed by evaluating that expression against a real
record. The UI marks fields like this (inferred) for exactly that reason: it
is one record's observation, not a promise. If your data is heterogeneous, one
sample will not tell you so.

The record underneath is pulled from the topic, not fabricated. You can draw a
different one, roll a fresh one from the schema, or type the case you are
actually worried about — the order with the null field, the quantity nobody
expected — and watch what your expression does to it. Whatever you put in
there, every downstream node reads the same record, so one story flows through
the whole chain.

Inference reads one execution path of one sample, so there are nodes where it
will be wrong and nodes where it has nothing to offer. Any node can declare
its output instead — a serde per side, and a subject if that shape is
registered. A node that declares one becomes a fresh start: the walk resumes
from it, and everything below reads what you stated rather than what was
observed. What you declare beats what was inferred, and both beat a guess.

It is deliberately not a free-form field list. Fields belong to schemas, and a
shape worth describing is usually a shape worth registering — so declaring a
node's output asks the same question binding a source does, and takes a
contract if one exists.

One thing the completions cannot warn you about: the values are runtime
objects, not the Java types their names suggest. A field the schema calls a
string does not arrive as a String — it arrives as org.apache.avro.util.Utf8,
a CharSequence wrapping the raw bytes. Equality still behaves, because SpEL
compares CharSequence content, so value.get('item') == 'Kettle' is true when
you expect it to be. But .contains(...) is a String method that
CharSequence does not declare, so it needs a .toString() first. It is the
kind of detail that lives in serde Javadoc rather than anywhere you would think
to look.

Why the loop can be trusted

All of which is worth exactly nothing if the answers differ from what the
topology does once it is deployed. A preview that disagrees with the runtime
teaches you something false and lets you find out at deploy time — worse than
showing you nothing at all.

That is not something you can check from the outside, so here is how we check
it. Every expression in the test suite runs through all three surfaces. Two you
have already seen: the editor's validator, and the preview under the sample.
The third is the topology itself — built by the same code that builds a
deployed one, running the same serdes, fed a record, and asked what came out of the
sink. That leg runs in-process rather than against a broker, which is the only
way to do it per expression at test speed, but nothing about the expression's
path is simulated.

All three have to agree: same success or failure, same output once serialized,
same verdict from a filter. The cases run across Avro, Protobuf, and JSON
Schema, and inside those, nested records, unions, arrays, enums, and logical
types — and the set only grows: every disagreement we find becomes a case
before it becomes a fix. They are usually small and specific, like arithmetic
on an Avro field that the editor flagged and the runtime handled perfectly
well.

It runs

Configuration, submit, and the deployed topology with live per-node numbers:

Setting a dead-letter topic in the stream configuration, submitting the topology, and watching per-node throughput badges appear on the deployed canvas.

Those badges exist because of a naming decision made much earlier. Runtime
metrics are tagged with operator names, so joining them back to the picture
means knowing what each operator is called — and Kafka's own answer,
KSTREAM-MAPVALUES-0000000003, is generated from graph position and shifts the
moment you insert an operator upstream. The names in this topology are derived
from the nodes instead, which is why the describe() output at the top of this
post reads order-total-enrich. That is worth more than legibility: those names
also land in JMX, in log lines, and in the internal topic names on your
cluster.

Read them carefully, though. The source reports an exact count; everything
downstream reports an attributed one, taken from the surrounding subtopology.
For a linear chain, the number is correct, but it cannot tell you which records
survived a predicate — the filter's effect shows up on the sink topic's
offsets, not on the node's badge.

What this is for

There is a stretch of work at the start of any pipeline that is mostly
questions: what is actually in these records, does this field mean what its
name suggests, what does my transformation do to the awkward ones. Answering
those in a project means writing an app to find out. Answering them here takes
the time it takes to type an expression, and the answers come from records that
are really on the topic.

The loop also explains an architectural choice. Most Kafka consoles are
readers: they ask the cluster questions — what topics exist, where are the
consumer groups, what do the metrics say — and never execute anything
themselves. Evaluating your expression against a real record and showing you
what came out is not a question you can ask Kafka. It means building the
topology and running it, so a Kafka Streams application deployed from this
console runs inside the console's own process.

Having everything in one place pays off after the deploy, too. The sink topic
is a topic like any other: the browser that showed you orders in the first
clip will show you orders-enriched, so you can read what your expression
actually produced — not just how many records got there. That is the round trip
closed: the schema you started from, the records you tested against, and the
output you caused, all reachable without leaving. Consumer groups, connectors,
and ksqlDB sit on the same sidebar for the same reason, and that is the
direction the rest of it keeps moving.

That is a genuine trade, not a free win: it puts your expressions in our JVM,
which is why the sandbox above came first. If you already have a console you
like for browsing a cluster, keep it — this one is shaped the way it is
because authoring needed something that could actually run what you wrote.

What stays out of reach

The derivation stops, and where it stops is the honest part.

At a join, we said nothing when this was published. A join emits when a
pair co-occurs — same key, both sides, inside the window — and inventing the
side that has not arrived asserts a match that may never happen. Since
0.10.0 the answer is two journeys instead of silence: each
input is evaluated on its own lane, #leftValue and #rightValue are bound to
their own branch's record, the joiner runs over both with the deployed
operator, and the nodes after the join get a chain like any other. What did
not change is the refusal to guess: a join whose second input cannot be
evaluated waits, and says so, rather than showing output for a pair that may
never exist.

One record at a time is not coverage. You can cycle through real records,
generate them from the schema, and write the awkward case by hand — but nothing
enumerates the shapes your stream actually contains, and the loop will never
tell you about the one you did not think to try. It is a probe, not a proof.

Protobuf values were opaque to all of this when this was published — and
that one was ours, not Kafka's. They arrive as DynamicMessage, which has no
get(String). It shipped in 0.10.0: a SpEL
MethodResolver gives get('field') the same meaning on Protobuf as on Avro
and JSON Schema, in the editor and deployed alike.

Those three were not the same kind of limit, and two have since moved.
Sampling one record at a time is not something a different tool fixes; it will
still be true in whatever you author with instead. Protobuf access was a gap in
what we had built, and it closed. The join was a decision — declining to guess
— and the decision stood; what changed is that a second lane, evaluated on its
own record, is not a guess.

What a tool can do is derive what is derivable, say nothing where nothing is
knowable, and let you run an expression against a record that actually exists
before you commit to it.

Try it, and tell us where it fits

This started with an orders topic and a transformation you could describe in one
sentence, with a build and a deploy standing between you and knowing whether
it was right. The four clips are that distance getting shorter: the schema
read out of the records, the expression checked against one of them before it
ships, the topology running with its own numbers on it.

What comes out is a Kafka Streams topology that runs as it stands. And if what
you wanted was a shape you now understand well enough to go and write in Java,
the value has to survive you doing that, and it does: the fields you confirmed
and the transformation you settled on do not care where they end up.

Where it stops being useful is the part we cannot answer alone: the pipelines
you would want to explore this way, and the ones where none of this would
help.

All of this is in the playground:
the orders topic with its registered schema, and the topology from these
clips deployed and running against it. One command, and you can take it apart
instead of taking our word for it.

Updates

2026-09-07 — when first published (2026-08-19), this post said Protobuf values were opaque to the expression editor and that we had not taught it to read them. That shipped in 0.10.0 on 2026-09-03: get('field') evaluates on Protobuf in the editor and deployed alike. The passage above was corrected to say so.
2026-09-08 — the same first version said the walk halts at a join and everything after it is left unevaluated. Since 0.10.0 a join is evaluated on two lanes and the nodes downstream get a chain; a join with an input that cannot be evaluated still waits rather than guesses. The join passage and the closing comparison were corrected to say so.

KafkaStreams.state() says REBALANCING. Every thread is dead.

alginte software — Sun, 02 Aug 2026 00:00:00 +0000

We found this one the embarrassing way: a seeded demo stream sat inREBALANCING for a day. Not failing — rebalancing. The UI said so. The Streams client said so. The only place the truth existed was a server log nobody was reading.

The trap

Deploy a Kafka Streams topology whose source topic doesn't exist , and here's the exact sequence (Kafka clients 4.x, but the behaviour is old):

The group leader's assignment fails withINCOMPLETE_SOURCE_TOPIC_METADATA; the member receives the error in its assignment.
The StreamThread logs MissingSourceTopicException, transitionsPENDING_SHUTDOWN → DEAD, and does not retry. This is deliberate — a missing source topic is not a transient condition Kafka Streams can wait out.
Crucially, the client-level state machine never follows. TheStreamsUncaughtExceptionHandler isn't consulted (the thread shut down; it didn't throw), so there's no PENDING_ERROR → ERROR transition.KafkaStreams.state() last saw a rebalance start, and that's where it stays.

The result: a KafkaStreams instance with zero live threads that reportsREBALANCING indefinitely. It will never process a record, never error, and never change state again.

Why every Kafka UI has this bug

If your tool renders KafkaStreams.state() — and that's the obvious, documented thing to render — you have this bug. The state enum simply has no value for "all my threads are dead but nobody told the coordinator layer."REBALANCING is the truthful answer to the wrong question.

The signal that does exist is one call away:metadataForLocalThreads() returns per-thread metadata including each thread's state. A client reporting REBALANCING whose thread set is empty — or whose threads are all DEAD — is not rebalancing. It's gone.

The fix, in two halves

Surface it. We derive the displayed state instead of trusting the raw one:

static KafkaStreams.State effectiveState(KafkaStreams.State state,
                                         Collection<ThreadMetadata> threads) {
    if (state == KafkaStreams.State.REBALANCING
            && (threads.isEmpty()
                || threads.stream().allMatch(t -> "DEAD".equals(t.threadState())))) {
        return KafkaStreams.State.ERROR;
    }
    return state;
}

Genuine rebalances have live threads (STARTING, PARTITIONS_ASSIGNED, …) and pass through untouched. In our end-to-end test, deleting a running stream's source topic flips the reported state to ERROR within seconds — where before it showed REBALANCING until someone read the log.

Prevent the common case. The most frequent way to hit this is a typo'd topic name at deploy time. Since a missing source topic is unrecoverable by design, we now validate every source node's topics against the cluster before building the topology, and fail the deploy with the missing names — one listTopics() round-trip. (Fail-open if the listing itself errors: a broker hiccup shouldn't block a deploy that would have succeeded; the state derivation above is the backstop.)

Neither half needs anything from the broker that isn't already public API.

Takeaways

KafkaStreams.state() is the state of the coordinator conversation, not the health of your processing. Dead threads don't move it.
If you're operating Kafka Streams with your own dashboards: alert on thread liveness (metadataForLocalThreads(), or the alive-stream-threadsmetric), not on state() != RUNNING.
If you're building a tool: derive, don't relay. The raw state is truthful and useless at the same time.

Both fixes shipped in Alginte 0.7.0. The stuck demo that taught us this now recovers in seconds — and deploying against a typo'd topic tells you the topic's name instead of miming a rebalance.

DEV Community: alginte software

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

First, the transform that may end the post here

Flatten after

Route by op

Survive the delete

And then the tombstone

The one that bites silently

Prove it before it runs

The short version

A Kafka Streams topology starts in a sandbox, not on your cluster

The expensive part is not the topology

A stack of your own, free to break

Carrying it across the border

What the sandbox does not prove

The order of operations

Monitoring Kafka Streams applications: the state a cluster view cannot show

Visibility into the stream's shape

Visibility into the stream's operation

What seeing it does not change

See what it is doing

The compiler was never what you wanted

Seven steps

Then you want to use a new field

What the seven steps buy

A tighter loop

What you give up

SpEL in Kafka Streams: from evaluator to editor

Why SpEL

Where the knowledge comes from

Completions from types, and from content

What an expression receives, and returns

Errors a user can act on

The tokenizer you cannot reach

What it costs

Is there a better way to do this?

Updates

Kafka Streams topologies you can draw and run

The data is already there, and so is its schema

What goes in the boxes

Drawing it, and watching where the type survives

The loop

Why the loop can be trusted

It runs

What this is for

What stays out of reach

Try it, and tell us where it fits

Updates

KafkaStreams.state() says REBALANCING. Every thread is dead.

The trap

Why every Kafka UI has this bug

The fix, in two halves

Takeaways

Flatten `after`

Route by `op`