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.

The Example Topology

Here is a real topology: four operators, reading orders and writing orders-enriched.

Sub-topology:
Orders-src (topics: [orders]) → order-total-enrich
Processor: order-total-enrich (stores: []) → big-orders-bigOnly
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.

Shape vs Behaviour

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.

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

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.

Sandboxed Expressions

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.

Building the Canvas

Four operators dragged out, wired, named, and pointed at orders:

• Source: Orders-src (topics: [orders])
• Processor: order-total-enrich (stores: []) → big-orders-bigOnly
• Sink: Big-Orders-sink (topic: orders-enriched) ← big-orders-bigOnly

Building the topology on a canvas involves 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. The pattern is simple: Type, look, adjust. Write a mapValues expression with completions drawn from the registry schema, watch 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. 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.

Validation & Testing

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:

1. The editor's validator
2. The preview under the sample
3. 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
• 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.

Purpose

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.

Architecture & Trade-offs

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 Support

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 Protob.