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

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.