Kafka Streams topologies you can draw and run

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.