DEV Community: alginte

Calling an external service from Kafka Streams: size the threads, budget every call

alginte software — Sat, 31 Oct 2026 17:58:38 +0000

You have a Kafka Streams application, and one step needs an answer from somewhere else. It might be an AI model's decision, a fraud score, an address to geocode or a status from a provider's API. The most direct way to write it is a blocking call inside a mapValues:

stream.mapValues(value -> {
    HttpResponse<String> answer = http.send(requestFor(value), BodyHandlers.ofString());
    return withAnswer(value, answer.body());
});

This is where most applications start. It is kept bare on purpose: no timeout, no retries, no error handling.

What max.poll.interval.ms bounds in Kafka Streams showed what such a call can do to a thread's place in the group, run by run. This post is the other half: what to do about it. A blocking call caps a stream thread at one record per call, so we start by sizing the threads for it. Then we give every call an upper limit, and keep the slow step from holding up the rest of the stream.

Size the threads for the call

Kafka Streams gives each input partition a task and spreads the tasks over its threads, as the previous post opens with. A thread processes its tasks' records one at a time, so a blocking call blocks the thread it runs on.

Our runs called a stub, a small HTTP server of ours that stands in for the service. It waits a set time and answers. Everything else was real: a broker, a Kafka Streams application, HTTP calls over the network, and real vulnerability records, CVEs (how we measured). The first run had three partitions and one thread, so three tasks took turns on it. The arithmetic is short: at 300 ms a call, a thread does 3.3 records a second, however fast records arrive. With 600 records arriving at 20 a second, we measured 3.2 to 3.4 a second, and the stub never saw more than one call in flight.

More partitions alone do not help. A repartition to 32 before the call gives the calling step 32 tasks, and so does an input topic of 32 partitions, which is what we measured. With the same single thread, it still did 3.3 records a second, one call at a time.

It needs threads as well: 32 stream threads (num.stream.threads), one per task of the calling step, counted across all the application's instances. At 300 ms a call, we measured 3.3 records a second with one thread, 26 with eight and 106 with 32. With 32, the stub saw 32 calls in flight at once. Of a backlog of 3,393 records, 3,247 were done after 35 seconds. The rest took until 50 seconds. The records were spread unevenly over the 32 partitions, so as the emptier ones ran out, their threads went idle and the fullest finished alone.

How many threads a stream needs follows from Little's law: calls in flight equal the input rate times the call time. What a number of threads can do, and how fast they clear a backlog, is then one division and one subtraction. With the numbers of our runs:

calls in flight = input rate × call time
                    = 25 a second × 0.3 s = 7.5, so at least 8 threads

capacity = threads ÷ call time
  8 threads = 8 ÷ 0.3 s ≈ 27 a second (measured 26.4)
  32 threads = 32 ÷ 0.3 s ≈ 107 a second (measured 106)

backlog drains = capacity − input
  8 threads = 26.4 − 25 = 1.4 a second
  32 threads = 106 − 25 = 81 a second

service paused 45 s = 45 s × 25 a second ≈ 1,100 records
  8 threads = 1,100 ÷ 1.4 a second ≈ 13 minutes
  32 threads = 1,100 ÷ 81 a second ≈ 14 s (measured about 15)

The pause is from our last run. We paused the service for 45 seconds, and the backlog grew to about 1,100 records.

Eight threads only just keep up. The threads beyond that are headroom. Headroom is what clears a backlog, whatever builds it: a pause in the service, as in our run, a slow minute of it, a spike in the input or a restart. Headroom also covers a slower service. At 600 ms a call:

  8 threads = 8 ÷ 0.6 s ≈ 13 a second, below the input
  32 threads = 32 ÷ 0.6 s ≈ 53 a second, above it

With 8 threads, the lag would grow by about 12 records a second. With 32, it would still drain.

Is the stream keeping up?

Once a stream runs, the consumer group's lag tells you whether that sizing still holds. With Kafka Streams the lag needs reading with care.

Operators watch a consumer group's lag: the LAG column of kafka-consumer-groups --describe, or a lag exporter's line on a Grafana dashboard. That lag comes from the group's committed offsets, and Kafka Streams commits only every commit.interval.ms, 30 seconds by default. So a stream that keeps up does not show zero lag. Between commits the committed offset stays put while records keep arriving, so the lag climbs. At each commit it drops. A stream that kept up with 25 records a second showed a lag climbing to about 750, which is 25 a second times 30 seconds. It dropped to almost nothing at each commit, every 30 seconds.

A dashboard reads this curve at its own interval. Read every 20 seconds, the dashed line, the same healthy stream jumps between about 230 and 750. A reading just after a commit could show almost zero, but most readings show hundreds.

The next chart adds a slow stream. It had a 300 ms call and received the same 25 records a second as the healthy one. It got through only 3.3 of them a second, one thread's worth at 300 ms a call. Its committed lag climbed to about 3,000 in 135 seconds and never came back down.

Lowering commit.interval.ms makes the teeth smaller, but each commit is work. The thread flushes its producer and its store caches and writes its offsets to the broker, and it processes nothing meanwhile. Raising it to commit less often has a cost of its own for slow calls, which What max.poll.interval.ms bounds in Kafka Streams measures.

Give every call an upper limit

Sizing the threads used the typical call time. The rules in the next section start from a different number: the longest a call can take, every retry included. We call it the call’s budget. Without one, no setting keeps a thread in its group. A single call that never ends outlasts any poll interval.

The budget is the whole call, not one setting. It is the time to connect, the wait for the answer, the time to read the answer's body, every retry, and the backoff between retries. A client that tries four times with a 5-second timeout and a second of backoff between tries has a budget of about 23 seconds, not 5.

A timeout on the request is not always that budget. We checked the JDK's own HttpClient, on Java 25, against our stub, with a 5-second request timeout. When the service sent nothing for 10 seconds, the call ended at 5.01 seconds. When it sent its headers at once and its body over 20 seconds, the call did not end. It returned normally after 20.05 seconds, with the whole body. The Javadoc says only that the timeout fires "if the response is not received within the specified timeout".

What ended that call was a deadline around the whole of it:

HttpResponse<String> callWithin(HttpRequest request, Duration budget) throws Exception {
    CompletableFuture<HttpResponse<String>> call =
            http.sendAsync(request, BodyHandlers.ofString());
    try {
        return call.get(budget.toMillis(), TimeUnit.MILLISECONDS);
    } catch (TimeoutException e) {
        call.cancel(true);
        throw e;
    }
}

The caller was free at 5.02 seconds. The cancel also closed the connection: the stub's next write, a second later, failed.

Retries go inside the same deadline. Each try gets what is left of the budget, so the budget stays one number.

With another client. We ran one client, one call per case, and no retries. Check two things. Does its timeout end a call whose body arrives slowly? If not, put a deadline around the call. Does cancelling stop the call, or only free the caller?

The arithmetic on the budget

A stream thread has to call poll() at least every max.poll.interval.ms, 300 seconds by default, or the group takes its tasks away in a rebalance. So two limits apply between two polls, and each is a number of calls times the budget:

round = tasks per thread × budget
poll's worth = max.poll.records × budget

A round is a call for each of the thread's tasks. One poll returns up to max.poll.records records, a poll's worth. How many of them the thread works through before it polls again depends on its history: while the service is fast, Kafka Streams lets it take more. A cap on max.poll.records bounds that, whatever the history. The consumer's Javadoc gives the setting this job: to bound "the maximum that must be handled within each poll interval". Keep both under half of the poll interval, 150 seconds by default. The other half is margin: when an instance goes away, its tasks move to the others, and their rounds grow.

For the stream we measured, 25 CVEs a second at 300 ms a call on 8 threads, with a 2-second budget:

round
  8 partitions = 1 × 2 s = 2 s, under 150
  32 partitions = 4 × 2 s = 8 s, under 150

poll's worth
  1,000, the default = 1,000 × 2 s = 2,000 s, far over 300
  75 = 75 × 2 s = 150 s, just half

So max.poll.records can be at most 150 s ÷ 2 s, 75. Set it there, or lower.

The bound assumes every call takes its whole budget. That is not rare: when the service slows down, it slows down for every call at once. And even at the typical 300 ms, the default 1,000 records take 300 seconds, the whole poll interval.

Tasks per thread is the calling step's partitions over its threads, all instances counted. Leave commit.interval.ms at its default. If you raise it above a round, keep the round under a third of the poll interval instead.

The table adds two more services. Their numbers are calculated, not measured: we did not run them. Their threads come from the typical call, by Little's law. Their two limits come from the budget. In the CVE column the rate, the call time and the threads are our runs, and the budget is an example.

	Fraud score	CVE severity, our runs	AI model decision
Records a second	200	25	5
Typical call	50 ms	300 ms	3 s
Calls in flight, rate × typical call	10	7.5	15
Threads, one task each	12	8	16
Budget, retries included	0.5 s	2 s	20 s
Round, with one task per thread	0.5 s	2 s	20 s
Poll's worth: `max.poll.records`, at most 150 s ÷ budget	300	75	7

Even the fast service needs max.poll.records cut, from 1,000 to 300. The AI model allows only 7 records a poll. That is the price of a 20-second budget, short of raising the poll interval.

Break either limit and the cost is not lag. The thread is evicted, its tasks start again from the last commit, and every call since then is made again. If the service charges per call, you pay each time. We ran three tasks on one thread with a 12-second call, against a poll interval scaled down to 30 seconds. The thread was evicted four times in 210 seconds and never committed, and the output held 17 records for 7 CVEs, where each CVE should give one. What max.poll.interval.ms bounds in Kafka Streams shows each bound run by run, and weighs the other levers, raising the poll interval among them.

Keep the slow step in its own stream

Threads are shared by all the steps of one application. So a fast step whose task lands on a thread with a slow one waits behind its calls. Two applications keep them apart. The first does the fast work and writes the records that need an answer to a topic. The second reads that topic, makes the call and routes the result. In our example, the first stream keeps the new CVEs that arrive without a severity score from their CNA, the organisation that assigned the CVE. The second asks the service for one.

The two-stream design.

The last run tested that, and the timeout, together. The first stream passed records on with no call. The second, with 32 threads, called the service. With records arriving at 25 a second, we paused the service for 45 seconds. With a 60-second timeout, the second stream simply waited out the pause, and its backlog grew to about 1,100. When the service came back, it caught up at about 100 a second in about 15 seconds. Nothing failed, and nothing rebalanced. With a 5-second timeout, 31 calls timed out together. The second stream went to ERROR, the client state that means it has stopped for good: Kafka Streams does not restart it, even after the service is back. In both runs the first stream never noticed.

The 5-second run stopped because of a default: a single call that throws sends the whole client to ERROR, with no retry, and a single connection error does the same. KIP-1033 added a handler in Kafka 3.9 that can change that policy. What to change it to, skipping the record, retrying it or sending it to a dead-letter topic, is a subject of its own. Until you have one, set the budget from what the service needs, not shorter.

Why not a separate worker

Another common way to handle a slow call is to take it out of Kafka Streams entirely. In our example, stream B would become a worker: a separate program of your own, built on a plain Kafka consumer rather than on Kafka Streams. It reads severity-requests, calls the service with many requests in flight at once, and writes the decided records to cve-decided.

Stream B needs a thread for every call in flight, because each call blocks its thread. A worker can use an asynchronous client instead: one thread sends many requests and handles each answer as it comes. For a very slow service that matters. At 25 records a second and 20 seconds a call, Little's law asks for 500 calls in flight: 500 threads and 500 partitions for stream B, a few threads for a worker.

In exchange, the worker has to do itself what Kafka Streams does for stream B, and all of it comes from the many calls it holds open. They finish in any order. So it commits an offset only when every earlier record of the partition is done, and keeps the order within a partition where that matters. And when it stops, or hands a partition to another instance, it has to finish or abandon the calls still open. Abandoned calls are made again by whoever reads the partition next.

It is also a second kind of application to watch. Its consumer group's lag shows like any other. But whatever shows you your streams, their state, threads and tasks, does not see the worker. Its health, its calls in flight and its failures need watching of their own.

Our call sits at the edge: stream B ends with it. A call in the middle of a stream, between a join and an aggregation, is harder to move out. The answers have to come back into the stream, joined to their records by key within a window. An answer later than the window needs a path of its own.

Two tools sit between stream B and a worker of your own: Confluent's Parallel Consumer and Responsive's async processor. We measured neither of them, nor a worker. The Parallel Consumer is the usual building block for a worker: more calls in flight than partitions, with order kept per key. The async processor keeps the call inside Kafka Streams and hands it to a thread pool, again with order kept per key.

The short version

A blocking call caps a stream thread at one record per call: throughput is threads over call time, up to the partitions.
Partitions alone do not help: a repartition without the threads to match only gives each thread more tasks.
Give every call a budget: the whole call, retries and backoff included. Without one, no setting is safe.
Keep a round, the tasks per thread times the budget, and a poll's worth, max.poll.records times the budget, under half of max.poll.interval.ms.
An eviction before a commit repeats the calls and writes duplicates.
With the defaults, one failed call stops the whole client.
Give the slow step its own stream, with the threads to match its partitions.

How we measured

We ran everything on one machine and one broker, against the stub, which waited a fixed time before it answered. A real service's time varies. Ours did not, so each run shows one latency at a time. The numbers describe the shape of the problem, not your cluster's. The records were real: newly published vulnerability entries, CVEs, from the National Vulnerability Database, replayed from one captured window. The runs used Kafka Streams 4.2.0 and the classic group protocol.

The stub, the replay script and the application that measured all of this are not published. If you want to run them against your own numbers, write to info@alginte.com and we will send them.

This post uses the NVD API but is not endorsed or certified by the NVD.

What max.poll.interval.ms bounds in Kafka Streams, measured

alginte software — Wed, 21 Oct 2026 16:46:42 +0000

Kafka Streams scales by dividing the work. Each partition of the input becomes a task. The tasks are spread over the stream threads of all the instances that share an application.id. Add an instance, or raise num.stream.threads, and the group hands the new threads some of the tasks.

The threads share no state, so they need no coordination among themselves. The consumer group coordinates them instead. It distinguishes a working thread from a stuck one by the poll: a thread has to call poll() at least every max.poll.interval.ms. If it does not, the group rebalances. It takes the thread's tasks away and hands them to the others.

A rebalance has a cost. Whatever a thread had done since its last commit is done again by the next owner of its tasks. A stateful task that moves to another instance cannot resume until its state stores are restored from their changelog topics. Kafka's docs name that restoration as the main cost of moving a task (Fault Tolerance). A standby replica shortens it. Another instance keeps a copy of the state up to date, and the task moves there. Standbys are off by default.

When you add an instance, the rebalance that follows is worth its cost. When a slow thread causes one, it is best avoided. This post is about the second kind.

Polling in time is easy while each record takes microseconds or milliseconds. A remote call changes that. Each record takes as long as the service takes to answer, or until a timeout cuts the call off. The thread waits either way. This is not yet the common case for Kafka Streams, and the standard advice is not to block a stream thread at all. On the other hand, we expect such calls to become more common. More teams now call external services on events as they arrive: an AI model's API above all, but also a scoring service, or anything else that answers in seconds.

A service's answer time also changes over the day. It answers quickly when it is quiet and slowly when it is busy. It is often busiest just when your stream has the most to send it. A slow service is a known hazard. A service that becomes slow is a different one. It hurts the threads that had been keeping up, not the ones that were already behind. We come to why later in the post.

The usual reading is that max.poll.interval.ms bounds one call. Keep each call under it, and the thread keeps its tasks. It bounds something larger. On Kafka Streams 4.2.0, we saw threads evicted from the group even though every call took well under the poll interval. Here is what we found, with the section that shows each:

The poll interval bounds more than one call. A thread with several tasks makes a call for each of them between two polls (the round).
A thread that had been fast was evicted again and again once its service slowed, at the default settings (a fast thread's service slows down).

Two things depend on the group, not the thread. A rebalance opened by another member can evict a thread whose settings are safe. And under Kafka's newer streams protocol, what an eviction costs depends on who else is in the group. Both are measured in a coming post.

We measured each case in a run, and predicted each run before it ran. Every run used a single machine with one broker and a stub HTTP service whose latency we set. Every run used the classic group protocol, Kafka Streams' default group.protocol. A mapValues called it and blocked until it answered. The use case is illustrative: a stream that sends each new security advisory, a CVE, to a service that scores it. The records were real, newly published CVEs, replayed from one captured window. The numbers describe the shape of the problem, not your cluster. The method, and the predictions that were wrong, are in How the numbers were made.

The stream thread's loop

Matthias J. Sax explains the stream thread's loop in his talk The Nuts and Bolts of Kafka Streams. From about nine minutes in, he covers the buffers, the bursts of records per task and why their size is adjusted. We build on his picture and follow one case: what that adjustment does to a thread whose every record waits on a slow call.

Between two polls, the thread has work that comes due. A commit is due once the commit interval has passed, and a poll before the poll interval runs out. A punctuation is due when it is scheduled. Checking for due work after every record would cost time. Checking too rarely would make that work late. So the thread works in passes of up to N records per task and checks after each pass. While nothing is due, it adds one to N, and the passes grow. When something comes due, it halves N, so the next check comes sooner. It is the rule TCP uses for its congestion window: grow slowly, back off fast. The rule assumes a record is cheap, microseconds or milliseconds. A remote call breaks that assumption. The rest of this post follows what happens then, and which settings keep a thread in its group.

A poll hands its records to queues, one for each task and input partition. When a queue holds more than buffered.records.per.partition records, 1,000 by default, its partition is paused (StreamTask). The thread stops fetching from it until the queue is back within the limit. It still has to poll in time. In a pass, the tasks take turns. Each runs up to N records from its queue, one after another, before the next task starts (TaskExecutor). With three tasks, A, B and C, and N at 2, a pass runs two records from each in turn: A, A, B, B, C, C. After each pass, the thread decides what comes next (StreamThread, lines 1298 to 1320).

The loop: what a stream thread does after each pass, in Kafka Streams 4.2.0.

after each pass:
  run any punctuation that is due
  commit, if commit.interval.ms has passed on the commit clock
  if the pass processed nothing:
      go back and poll
  else if the time since the last poll is past half of max.poll.interval.ms:
      halve N, and go back and poll
  else if the pass committed or punctuated:
      halve N, and run another pass
  else:
      add one to N, and run another pass

The figure below follows the loop for three tasks at the default settings. Each call takes 40 seconds, longer than most, so that every pass is wide enough to see. Every pass takes longer than the 30-second commit interval, so every pass ends in a commit. Each commit halves N, and N never drops below 1, so each pass runs one record per task. The first pass ends at 120 seconds, short of the half-way mark, and the thread runs another. The second ends at 240 seconds, past the mark, and the thread polls with 60 seconds to spare.

Almog Gavra's Kafka Streams 101 on configs names the half-interval check. Two facts the rest of this post leans on are less known. N starts at 1 when the thread starts (line 877) and is never reset. It is carried from one poll to the next and, as we will see, across a rebalance. And a pass processes nothing only when none of its tasks has a record it can process. With a slow call and a backlog there always is one, so only the half-interval check sends the thread back to poll.

This is the default loop of 4.2.0. A second one, with processing threads, sits behind an internal flag that defaults to false. The code itself describes it as incomplete, and nothing here applies to it.

The unit is the round

A thread polls only after a whole pass, so what has to fit inside max.poll.interval.ms is not one call. It is at least one record of each of the thread's tasks, one after another. We call that a round: with T tasks on a thread and a call of L seconds, about T x L. A. Sophie Blee-Goldman put it in one sentence in Kafka Streams 101: Optimizing your apps: a thread makes progress as long as it processes one record per task within the poll interval.

The round grows with the tasks on a thread, as the figure below shows. An input topic of 32 partitions gave a single thread 32 tasks, and a round of 32 calls. With 32 threads, each had one task, and a round of one call.

In the figure above, every round fits the poll interval with room to spare, so a slow call costs only throughput. A round longer than the poll interval costs more. To show it, we set up a case on purpose: three tasks on one thread, a 12-second call, and a 30-second poll interval, so a round takes 36 seconds. The figure below draws its first pass. Every call fits well inside the poll interval, even inside half of it. The round does not, and at 30 seconds the client leaves the group.

We stopped the run after 210 seconds. In that time the thread was evicted four times and never committed. Each CVE should give one output record, but the thread had written 17 records for only 7 CVEs, the first few of each partition, again and again. Left running, it would have written more records for the same 7, and never reached the rest of the backlog.

The 30-second poll interval is the default of 300 scaled down tenfold, only to keep the run short. Later sections call these runs the scaled runs. At the default, the same case is thirty tasks and a 12-second call. Thirty tasks is what the default of one thread gets from a topic with thirty partitions, and an AI model writing a long answer can take 12 seconds. Or it is three tasks and a 120-second call. A call can take that long when the service is overloaded, or when the client retries with backoff. Several later runs use a 120-second call on a single task. What counts is the round, and ten tasks at 12 seconds make the same 120-second round. One task is simply easier to follow.

The commit interval is the other clock

With slow calls, leave commit.interval.ms at its default. Raising it saves little, and it opens a way to be evicted. The mechanism still matters, because a pass without a commit is also how a later case ends in an eviction: a thread whose service slows.

The round is not the whole story. How many rounds a thread runs before it polls depends on commits, as the loop shows. A commit is due when commit.interval.ms has passed on the thread's commit clock, which starts at the last commit (maybeCommit). A pass that commits halves N, the number of records each task runs in a pass. A pass without a commit grows it.

The defaults are a 30-second commit interval against a 300-second poll interval. Every pass longer than 30 seconds then commits, so N stays at 1, and a thread whose round fits the poll interval polls in time. That holds for a latency that has been steady since the thread started, and in a group with no rebalance under way. The first condition comes back below, and the second in a coming post. We checked it without scaling: one task, a 120-second call, 15 minutes. The thread committed after every record and never rebalanced, and the output held 7 records of 7 CVEs.

Each commit costs work: the thread flushes its state stores and writes its offsets, and processes nothing meanwhile. Raising commit.interval.ms saves some of that, which can buy throughput. With calls that take seconds, the saving is small, and the cost is what follows.

Raise commit.interval.ms above the length of a round, and a pass can end with no commit due. We ran the same one-task, 120-second call with both intervals at 300 seconds. The first pass after the poll ended at 120 seconds. That was not past the half-way mark of 150, and no commit was due. So N went to 2, and the next pass took two records, another 240 seconds. The thread could poll again only at 360 seconds, past the poll interval. A rebalance followed at 363.8 seconds, and the output held 7 records for 4 CVEs.

We call the range this round falls in the band: longer than a third of the poll interval, shorter than half. A round in the band ends its first pass before the half-way mark, so the thread does not poll. With no commit due, N goes to 2, and the second pass takes two rounds. The thread can poll only after three rounds, and three rounds run past the interval. A round under a third survives, because three rounds still fit. A round over half survives too, because its first pass crosses the half-way mark and the thread polls.

It is the client that leaves. When the poll timer expires, its own heartbeat thread sends the leave, and the processing thread learns of it at its next commit.

Runs on both sides of the band agree. In the scaled runs, the commit interval was left at 30 seconds against a 30-second poll interval, so commits were rare by accident. A round of 12 seconds (0.4 of the poll interval) was evicted and wrote 96 duplicates in 256 records. Rounds of 9 seconds (0.3) and 17 seconds (0.57) were not. One case needs care even at the defaults. A round just under half polls only after two rounds, seconds before the poll interval runs out. We measured the band with the commit interval equal to the poll interval. Other settings above the round follow from the loop, not from a run.

The same clock explains a smaller oddity. Commits land later than the commit interval says: about 35 seconds apart with a 300 ms call, and 35 to 40 with a 3-second one in a scaled run. A commit waits for the pass in progress, and between commits the passes grow by a round each. Apurva Mehta and A. Sophie Blee-Goldman's exactly-once guide calls the commit interval a lower bound, and this is why.

After an eviction

Go back to the band run of the last section: one task, a 120-second call, and both intervals at 300 seconds. It was evicted, and rebalanced at 363.8 seconds. Our first reading of the loop said it would then be evicted every time it rejoined, since it would start the same way again. It was not. After rejoining, it kept its place, because a rebalance is not a fresh start. One thing survives it that a restart would reset, and one thing restarts.

N survives. The failed commit that reveals the eviction throws from the commit check, before N is adjusted, and N is reset only when the thread starts. So the first pass after rejoining took two records, 240 seconds, crossed the half-way mark and polled in time.

The commit clock restarts. The clock counts the time since the last commit. Once it passes the commit interval, the next commit check commits, and restarts the clock even when there is nothing to commit (line 1881). After the rejoin, the task needs a moment to start, and the thread runs a few passes that process nothing. The first of them finds a commit overdue, commits nothing and restarts the clock. So the next real commit falls due 300 seconds after the rejoin. The log does not show those empty passes. We inferred them because nothing else fits the timings, and we predicted the next eviction from them.

We ran the same settings again for 1,500 seconds, and they did what that predicted. The thread rebalanced at 363.4 and 1,203.9 seconds, committed once between them at about 725, and wrote 12 records for 7 CVEs. A thread in the band is evicted again, but not every time it rejoins. When depends on where N and the commit clock stand. Each eviction costs the calls made since the last commit, made and written again.

When a fast thread's service slows down

That N survives a rebalance has a consequence that matters more than the band, and it needs no setting raised. N has no upper limit in the code. It grows by one after every pass that neither committed nor punctuated, however few records the pass held. A stream that keeps up therefore carries a large N. Ours, at 25 records a second with the service answering at once, reached 221 in two minutes. A pass runs at most N records per task, but only the records waiting in the queues. So a large N means a pass takes everything waiting.

A commit halves N only when the check falls on a pass that held records. On an idle pass it leaves N alone, and most passes of a stream that keeps up are idle. In another run of the same two minutes, one check did land on a pass with records, and N ended at 124. Kafka logs N at DEBUG in every pass, as "iterations", which is where these numbers come from (line 1283).

Then we slowed the service to 10 seconds a call, a thirtieth of the poll interval, with 40 records waiting and every setting at its default. The thread took the 40 as one pass of 400 seconds, with no look at the clock inside it. The poll timer expired at 300. The pass's commit, the first attempt in 415 seconds, failed. N was 222 after the rebalance as before it, so the next pass was the same records again. It was evicted every time it rejoined: 401, 811 and 1,222 seconds after the slow-down, and a fourth was under way when the run ended. In all, 155 records were written for 41 CVEs, 154 of them from calls of 10 seconds, and not one was committed.

The same 40 records at the same 10 seconds were no trouble for a thread that starts on them. N went 1, 2, 1, 2, with a commit after every second pass. There was no rebalance, and all 40 were done in about 400 seconds. The two threads differ only in their history. No setting states it, and nothing about the slow call shows it.

Ten seconds after an instant answer is an extreme slow-down. We chose it so that one run would be short and leave no doubt. The arithmetic of the loop puts the line much lower. With N at 222 and that many records waiting, a pass is 222 calls back to back. It has to end within the 300-second poll interval, so each call can take at most 300 ÷ 222, about 1.35 seconds. Any slower, and one pass outlasts the poll interval. The 222 is measured. The 1.35 seconds is only that division: we did not run a slow-down to it.

One more condition belongs here. We placed the backlog whole: the thread was held in one call while the 40 arrived, so that its next poll returned all of them. A backlog that builds by itself reaches the thread one poll at a time. Each pass then takes what arrived during the one before. When records arrive faster than the service answers, each pass is longer than the last. Arriving more slowly is not enough on its own: a backlog already waiting is taken in one pass, up to max.poll.records of it, and that pass still has to fit the poll interval. This follows from the loop, and we did not run it either.

The thread does not get out by itself. Each time it rejoins, N is still 222, the pass is the same 40 records, and the commit that would halve N is the one that fails. Only a new thread starts N at 1 again. Restarting the instance gives one, and so does replacing the thread at runtime with removeStreamThread() and addStreamThread(). The thread that started on the backlog shows what follows. A restart redoes the uncommitted records once. It is a cure, not a prevention: once the service is fast again, N grows back, and the next slow-down sets the same trap. To stay out, cap max.poll.records, so that a poll's worth of slow calls fits inside the poll interval. It saved this thread with nothing else changed, as the levers below show.

The levers, and what each costs

max.poll.interval.ms. Raising it lets a longer round fit, so a thread that was being evicted keeps its place and its calls are not made again. It adds no throughput: a stream that was being evicted falls behind quietly instead. And it has two costs.

The first: a thread that is alive but stuck, on a call with no timeout, keeps its partitions until the poll interval runs out. A process that dies is still noticed sooner, by the session timeout, because heartbeats come from a thread of their own.

The second: with the classic protocol, the default, a rebalance waits for every member to rejoin, up to the poll interval, as Blee-Goldman's guide to rebalancing describes. A busy thread rejoins only after the pass that takes it past half the poll interval. So with a 30-minute poll interval, a new instance can wait 15 minutes for its tasks, and longer by however long that pass takes. That follows from the loop, and we did not measure it. Until Kafka 2.3, Streams set this interval to effectively infinite. KIP-442 returned it to the consumer's default.

commit.interval.ms. Lowering it means more commits, and each one is a flush and an offset write during which the thread processes nothing. Raising it above the length of a round opens the band from a third to a half of the poll interval. Keep a round under a third then.

max.poll.records. Streams sets it to 1,000, and every run but one left it there. It caps what one poll hands the thread. For a thread whose N is larger than that, it caps the pass. We ran the fast stream that slows once more, with nothing changed but this setting, at 20. The poll returned 20 of the 40 waiting records, and the pass was 200 seconds. Its commit went through with 100 seconds to spare and halved N, and the other 20 went the same way. There was no eviction, and each record was done once, where at the default it had been evicted every time it rejoined.

KAFKA-14382 names lowering it as a workaround that shortens the processing loop. This is that, measured for one value. By the loop, the rule is that a poll's worth of calls has to fit the poll interval: max.poll.records times the call. What it costs a stream while it is fast, more polls for the same records, we did not measure.

More instances. They add threads, and threads add throughput when a round fits. They do not rescue a round in the band. A coming post shows that a second instance made the repeated work worse.

For a long call, then, these settings keep a thread in its group. Start with a timeout on every call, retries included. The other settings rest on it: without a timeout, no record has an upper limit, and no setting is safe. Keep commit.interval.ms at its default, below a round, so every pass commits and N stays at 1. Give each thread few enough tasks that a round fits well inside the poll interval. Cap max.poll.records so that a poll's worth of calls at the timeout fits too. That cap is what kept the thread whose service slowed. Raise max.poll.interval.ms last, for the costs above.

What we did not measure

A single machine with one broker, and a stub whose latency has no tail. A round just under half the poll interval has no margin for one.

A latency that changes, beyond one case. Every run but two held the call's latency steady from the start. The fast stream that slows was run twice, at 10 seconds, on one task, with its backlog placed whole: once at the default max.poll.records, and once at 20. A slow-down that comes gradually, a backlog that builds by itself, a service that recovers while a pass is under way, and other latencies are readings of the loop.

Exactly-once. Every run was at-least-once. A transactional commit takes another path through the broker, and the commit interval's default is different.

Kafka Streams 4.2.0 only. In that version, restoration runs on a separate state-updater thread by default, which is one way the loop can differ from older versions. We make no claim about any version whose source we have not read. Nor do we claim anything about the loop with processing threads, which 4.2.0 keeps switched off.

And one thing we saw and have not explained. In the run of The unit is the round, three tasks at 12 seconds against a 30-second poll interval, the rebalances came 36 seconds apart. Once, the gap was 72. If you know why, or have seen the same on your own cluster, tell us in Discussions. We will add what we learn to this post.

How the numbers were made

Every run was written down as a prediction before it ran. Its result was added after, without editing the prediction. The early runs were predicted from a model of the loop, the later ones from its source.

Several predictions were wrong, two of them about the mechanism. The first model, one round per poll, predicted that one task per thread would make a 12-second call safe against a 30-second poll interval. It was evicted. Reading StreamThread is how we found the adaptive pass. Sax's talk, which we found afterwards, explains it too. The prediction for the 120-second call with rare commits rested on a false detail, that the first pass would commit, so its eviction came earlier than predicted. Two more missed a number: commits predicted 30 seconds apart came 35 apart, and one end count was an arithmetic slip.

A note on names, for readers who open the source. The field that holds N is called numIterations. Kafka's thread metrics count an iteration as one poll and all the processing after it, which can hold several passes. We say pass to keep the two apart.

The stub, the replay script and the application are not published. If you want to run them against your own numbers, write to info@alginte.com and we will send them.

This post is one of a few on calling a service from Kafka Streams. A coming one starts from the timeout on every call: how to make sure a call is really cut off, and the arithmetic that then says which settings are safe. To get it, follow the blog's RSS feed or @alginte_sw on X, where we post each one.

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

Calling an external service from Kafka Streams: size the threads, budget every call

Size the threads for the call

Is the stream keeping up?

Give every call an upper limit

The arithmetic on the budget

Keep the slow step in its own stream

Why not a separate worker

The short version

How we measured

Related reading

What max.poll.interval.ms bounds in Kafka Streams, measured

The stream thread's loop

The unit is the round

The commit interval is the other clock

After an eviction

When a fast thread's service slows down

The levers, and what each costs

What we did not measure

How the numbers were made

Related reading

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`