How eight single-member consumers, each fetching 4,742 partitions in one request, brought a ten-broker Kafka cluster to a standstill one broker at a time, why it only started after an upgrade from the 2.x line, and why the brokers’ own latency metrics never noticed.
TL;DR. A very wide consumer fetch parks in
DelayedOperationPurgatoryas a singleDelayedFetchregistered under every partition it covers. Every produce, follower fetch and consumer fetch that completes on any of those partitions callscheckAndComplete, which takes that one operation’sReentrantLockand runs a completion check proportional to the fetch width. Since Kafka 2.7 that lock acquisition blocks instead of trying and moving on. With 864 watched partitions on one broker, 30 of 36 request-handler threads queued behind one lock, the request queue hit its cap, and the network threads of the busiest listener stopped reading sockets. Broker-measured latency stayed low because time in an unread socket is not measured. A producer that awaited one message at a time under a per-batch timeout turned every slow minute into cancelled and re-sent batches, feeding the convoy. Stopping the eight consumers gave 2.8× throughput at a fifth of the latency within three minutes.
The setup
The system is an event pipeline for high-volume financial-style transactions. An application writes a transaction to its database and publishes an event to a message queue. A publishing service consumes that queue in small batches and produces each event into a Kafka topic with 105 partitions on a ten-broker cluster, Kafka 3.7.0 in KRaft mode, deployed with an operator on Kubernetes with two brokers per physical node. Downstream, several hundred consumer groups read the resulting topics: analytics, rule engines, exports, and a replication job that copies production topics into another environment.
Three details of that setup turned out to matter more than everything else combined:
- The replication job was eight consumer groups, one member each, subscribed by wildcard to 285 topics and therefore 4,742 partitions. Each member’s fetch to a given broker spanned every partition that broker led.
- The publisher sent one message per produce request, awaited each delivery report before sending the next, and cancelled the entire batch on one shared timer.
- The cluster had moved from a Kafka 2.x release to 3.7.0 three months earlier.
The symptom, and why the metrics lied
From the publisher’s point of view, produce calls started taking 30 to 240 seconds and failing.
From the broker’s point of view, nothing was wrong: Produce total time p99 never exceeded
about 7.5 seconds during the worst episodes, and the request-handler idle ratio looked fine. For
five days the incident was debugged as a client-side regression.
The gap between those two views is the first lesson of this incident, and it is structural. Kafka’s request metrics start the clock when a network thread has read a complete request off the socket and enqueued it. If the network thread is not reading, the request is not in any metric. Meanwhile the handler idle metric on a broker with combined broker-and-controller roles is unreliable because of a JMX bean collision between the two request pools, so the pods that were actually stalling reported idle ratios above 1.0.
The metric that did show it, in every episode, was
kafka.network:type=RequestMetrics,name=LocalTimeMs,request=FetchFollower: replica-fetch local
time rising from 1 to 2 ms to 10 to 90 ms on one broker at a time, while the follower-fetch rate
collapsed, handler CPU went down and the Produce purgatory size climbed. Followers touch every
partition, so they hit the mechanism first. The second was per-processor
NetworkProcessorAvgIdlePercent: the eight processors of the external TLS listener at zero idle
while the internal listener’s processors were fine.
Why “idle at zero” meant blocked, not busy. A busy network thread has a high
io-ratioplusio-wait-ratio. A blocked one has both near zero: it is parked on a queue put, doing no I/O and not waiting on select either. On the stalled brokers the external listener’sio-ratio + io-wait-ratiocollapsed to about 0.1.
The mechanism
1. A convoy, not a deadlock
In a deadlock nothing moves. Here everything moved, through one door. One handler held a lock
while it did a long piece of work. Every handler that needed the same lock parked behind it.
When the holder released, the next one took the lock and did the same long work. The pool never
stopped; it just spent almost all of its time queuing. Three SIGQUIT thread dumps of the
stalled broker, taken about ten seconds apart, showed 33, 32 and 30 of the 36 data-plane
handlers parked on the same lock object each time.
2. How a fetch parks in purgatory
A consumer fetch that cannot be answered immediately, because fetch.min.bytes is not yet
satisfied, becomes a DelayedFetch in the broker’s DelayedOperationPurgatory. The purgatory
must wake it the moment data arrives on any partition in the request, so it registers the same
operation object in a Watchers list keyed by TopicPartition, once per partition in the fetch.
A fetch covering 864 partitions on this broker is one object under 864 keys.
Whenever a handler finishes work on a partition, appending a produce, serving a follower fetch,
or reading for a consumer, ReplicaManager calls purgatory.checkAndComplete(key) for that
partition. That walks the partition’s watch list and, for each operation, calls
safeTryComplete(), which takes the operation’s lock and runs tryComplete(). For a
DelayedFetch, tryComplete iterates every partition in the fetch to compute accumulated
bytes, and if the threshold is met it calls onComplete, which reads the log for every partition
and builds the response, still under the same lock.
Here is the reconstructed stack the parked handlers shared, abridged from the dumps:
"data-plane-kafka-request-handler-12" #… daemon prio=5 WAITING (parking)
at jdk.internal.misc.Unsafe.park
at java.util.concurrent.locks.LockSupport.park
at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquire
at java.util.concurrent.locks.ReentrantLock.lock
at kafka.utils.CoreUtils$.inLock
at kafka.server.DelayedOperation.safeTryComplete <-- 30 of 36 threads here
at kafka.server.DelayedOperationPurgatory$Watchers.tryCompleteWatched
at kafka.server.DelayedOperationPurgatory.checkAndComplete
at kafka.server.ReplicaManager.$anonfun$addCompletePurgatoryAction / tryCompleteActions
at kafka.server.ReplicaManager.appendRecords // or fetchMessages, for the follower/consumer variants
at kafka.server.KafkaApis.handleProduceRequest
at kafka.server.KafkaRequestHandler.runThirty of these per dump, three dumps, three distinct lock objects across the ten-second span as
one fetch completed and the next one from the same consumer parked. The single thread that was
not waiting was inside DelayedFetch.tryComplete or onComplete.
3. From parked handlers to a frozen listener
Handlers are the only threads that dequeue from RequestChannel. With thirty of them parked, the
queue reaches queued.max.requests (500 here) within minutes. Network processors block on the
queue’s put and stop calling select. Kafka assigns processors per listener, so the listener
carrying the most requests fills the queue first and blocks first. On this cluster that was the
external SASL_SSL listener the publishers used; in-cluster clients on the plaintext listener
barely noticed.
4. Why it started after the upgrade
The replication consumers had run in this shape for months on the 2.x cluster. What changed is
how DelayedOperation behaves when a handler finds the lock already held. Abridged from the two
sources:
// Kafka 2.5 / 2.6 — DelayedOperation.maybeTryComplete (abridged)
private[server] def maybeTryComplete(): Boolean = {
var retry = false
var done = false
do {
if (lock.tryLock()) { // non-blocking
try { tryCompletePending.set(false); done = tryComplete() }
finally { lock.unlock() }
retry = tryCompletePending.get()
} else {
// someone else holds it: leave them a note and move on
retry = !tryCompletePending.getAndSet(true)
}
} while (!isCompleted && retry)
done
}
// Kafka 2.7 → 3.7 — DelayedOperation.safeTryComplete (KAFKA-8334)
private[server] def safeTryComplete(): Boolean = inLock(lock)(tryComplete()) // blocksKAFKA-8334 fixed a real bug: with the
try-and-flag scheme, a completion could be missed when the flag was cleared at the wrong moment.
The fix is correct. It also changes the cost model: a handler that used to skip a held lock now
waits for it, and the wait includes the holder’s tryComplete over every partition in the
operation. For ordinary fetches of a few dozen partitions nobody notices. For one fetch of 864
partitions, watched under 864 keys, with a produce rate of thousands per second onto those
partitions, the pool collapses into a queue.
5. The selector: which broker stalls
The convoy forms on whichever broker leads the most of the partitions the wide consumers read. Balanced across ten brokers, each leads about 470 of the 4,742, and the lock’s arrival rate stays under what one thread can serve. The September hardware migration, a node retirement, several large reassignments and finally a storage-controller failure kept concentrating leadership on one broker at a time. The stall moved five times in eight days, and the ranking of replicated partitions led per broker predicted the ranking of stall severity exactly.
Blocked: client-listener processors idle 0.00, 30 handlers parked · Elevated: FetchFollower local p75 17–47 ms · Clean: ≤ 2 ms
6. The amplifier: the producer
A slow broker explains slow produces. The publisher’s design explains why slow produces became eight days of failed batches and a growing backlog. Its produce loop, in shape:
// one timer for the whole batch, sequential single-message produces
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(batchBudget)); // 120 s, later 240 s
foreach (var msg in batch)
{
await producer.ProduceAsync(topicPartition, msg)
.ContinueWith(r => RecordResult(r, msg.Id), cts.Token); // token cancels the continuation
}Four consequences follow directly:
- Budget per batch, latency per message. A 60-message batch fails whenever
60 × per-produce latency > budget, even if every single produce succeeds. At 2 s each that is exactly 120 s.
- The real error is unobservable. The token cancels the continuation, not the produce. The
delivery report, and with it Local: Message timed outor the broker’s error code, is never read. The log says only that a timer fired. Five days of debugging went into the wrong layer because of this.
- Duplicates. The in-flight message stays in librdkafka’s queue and may still be delivered; the batch is re-queued and the same message produced again. On the worst day, the ratio of queue deliveries to unique messages was about 1.74.
- Load. Every failed batch is re-produced in full, one request per message, onto the very
partitions under the convoy. The publisher alone accounted for about half the cluster’s
produce requests at 1.02 messages per request. librdkafka cannot coalesce what it never holds:
with one outstanding message per loop, linger.mshas nothing to batch.
The fix on the producer side is the standard shape, and it is worth spelling out because the
first version of the rewrite kept two of the old bugs. Produce everything in the batch without
awaiting each message, let each delivery handler complete its own TaskCompletionSource, await
the batch once with a bounded wait, and decide per message from the delivery report:
var pending = batch.Select(msg =>
{
var tcs = new TaskCompletionSource<DeliveryReport>(TaskCreationOptions.RunContinuationsAsynchronously);
try
{
producer.Produce(topicPartition, msg, report => tcs.TrySetResult(report)); // non-blocking
}
catch (ProduceException<K, V> e) // Local: Queue full, serializer errors
{
tcs.TrySetResult(DeliveryReport.Failed(e.Error)); // per-message outcome, never abort the batch
}
return (msg, tcs.Task);
}).ToList();
var all = Task.WhenAll(pending.Select(p => p.Task));
if (await Task.WhenAny(all, Task.Delay(batchBudget)) != all)
{
// budget exceeded: mark still-pending items failed with a *specific* reason;
// do not cancel their continuations
}
foreach (var (msg, task) in pending)
Ack(msg, task.IsCompletedSuccessfully && !task.Result.Error.IsError, task.Result?.Error); // log Error.Code and PersistenceStatusTwo producer settings finish the job. message.timeout.ms should sit below the batch budget so
librdkafka fails stragglers with a report before the application’s timer does. And linger.ms
has to be raised deliberately: at roughly 430 messages per second per host spread over ten
brokers, the default 5 ms almost never coalesces, and the wire still carried 1.06 messages per
request after the rewrite. A 100 to 200 ms linger buys a four- to nine-fold reduction in request
count for a latency cost no analytics consumer will notice. Turn on enable.idempotence so that
the retries that remain cannot duplicate.
The experiment
By the evening of the worst day, leadership had been re-balanced and no broker configuration had changed for two hours. That made stopping the eight replication consumers a clean test: if the convoy was the cause, the cluster should improve immediately, everywhere, and in real deliveries rather than retries.
The mirror side of the same evening is instructive too: raising the publisher’s instance count from 11 to 20 and then to 30, an hour earlier, produced no measurable throughput and slightly higher broker latency. More senders into the same full queue. Ninety handlers behind one lock are still one lock.
Timeline, compressed
What I would tell another Kafka operator
Consumers
- Budget partitions per consumer member, not per topic. Any group that would read more than a few hundred partitions per member must be sharded before it is deployed. Treat it like replication factor: a reviewed number.
- Shard replication jobs across many members with the cooperative-sticky assignor and static
membership (group.instance.id), so a member restart does not trigger a full rebalance and a fresh full-width fetch. With 16 members the widest fetch on any broker here would be about 55 keys.
- Split wildcard subscriptions by topic family. A wildcard that silently grows with every new topic is how a fetch width becomes unknowable.
- A consumer that is millions of messages behind should be completing fetches immediately. If it
parks in purgatory at all, its fetch.min.bytes/fetch.max.wait.msare wrong for its width. Capfetch.max.bytesandmax.partition.fetch.bytesso the response built under the lock is small.
- Watch for consumer recreation. Two of the eight groups here recreated their consumer every one
to two minutes; every recreation is a new fetch session and a full, non-incremental fetch, with
the broker’s session cache already full at max.incremental.fetch.session.cache.slots=1000.
- Prefer a purpose-built replicator (MirrorMaker 2 with tasks.maxper topic family) over a hand-rolled single-consumer copy job. Replication tools partition the work by design and expose lag and throughput per task.
Brokers
- Request quotas for wide principals. A request_percentagequota on the replication principal would have throttled one badly shaped client instead of starving everyone behind the shared handler pool.
- Keep leadership balanced automatically and alert on it: PreferredReplicaImbalanceCount > 100, or any broker leading more than 1.3× the average. Never leave a broker at 1,600 leaders for hours because the rebalancing tool’s self-healing is broken.
- Rack awareness (broker.rack= physical node) when running two brokers per host. One controller failure here turned into 14 offline partitions and 140 under min-ISR because both replicas of a partition shared a host.
- Do not treat num.io.threadsorqueued.max.requestsas the fix for a convoy. More threads behind one lock are more parked threads. Check release notes after 3.7 for purgatory changes before treating an upgrade as a fix, and know that a correct upgrade can change the cost of an existing pattern.
- Isolate broker CPU from co-tenant workloads (requests, ideally Guaranteed QoS) and keep reassignments out of peak hours with two approvers. Several evening episodes here were timed by a co-located workload’s CPU bursts and by rebalances approved during the incident.
Producers
- Never await one message at a time under a batch-wide timeout. Produce with delivery handlers, await once, decide per message.
- Always read the delivery report. The error you are not logging is the one you will spend five days looking for elsewhere.
- Set message.timeout.msbelow any application budget so librdkafka fails stragglers with a reportable reason before your own timer does.
- Raise linger.msdeliberately and enable idempotence. At low per-connection rates the default linger coalesces nothing, and every retry without idempotence is a potential duplicate.
- Retries are load. Back-pressure and pausing replay jobs when produce latency rises beat re-sending whole batches into a struggling broker.
Detection: PromQL that would have caught it in minutes
# Earliest and most specific: follower-fetch local time on any broker (normal 1–2 ms)
histogram_quantile(0.75, sum by (pod, le) (rate(kafka_network_requestmetrics_localtimems_bucket{request="FetchFollower"}[3m]))) > 10
# A blocked listener: its processors' idle collapses to ~0 while handler CPU also falls
min by (pod, listener) (kafka_network_processor_idle_percent) < 0.2
# Request queue at cap (unreliable on combined broker-controller pods: bean collision reports 0)
kafka_network_requestchannel_requestqueuesize >= 0.9 * 500
# Produce purgatory climbing on one pod — the companion signal in every episode
kafka_server_delayedoperationpurgatory_purgatorysize{delayedOperation="Produce"} > 400
# The selector: leadership concentration
kafka_controller_kafkacontroller_preferredreplicaimbalancecount > 100
max(kafka_server_replicamanager_leadercount) / avg(kafka_server_replicamanager_leadercount) > 1.3
# The lock holder's shape: partitions per consumer member (kafka_exporter)
count by (consumergroup) (kafka_consumergroup_lag) / on (consumergroup) sum by (consumergroup) (kafka_consumergroup_members) > 300When the first of these fires, take three thread dumps ten seconds apart on the affected broker.
If the data-plane handlers are parked in DelayedOperation.safeTryComplete under
Watchers.tryCompleteWatched, it is this mechanism, and the immediate mitigation is to move
leadership off that broker or pause the widest consumer.
Closing
Nothing in this incident was exotic. A replication job with an unusual but legal shape, a Kafka fix that closed a real bug, a hardware migration that moved leaders around, and a producer written the way most first Kafka producers are written. Individually, each had been fine for months. Together, with the lock semantics of Kafka 2.7 and later, they formed a convoy that no single dashboard showed. The two things I would want any team running Kafka to take from it are: partitions per consumer member is a capacity number, and read your delivery reports.
Mechanism confirmed from three thread dumps of the stalled broker, per-episode Prometheus
metrics, broker logs, and the DelayedOperation.scala source for Kafka 2.5.0 and 3.7.0. Numbers
are as measured at the time. Kafka internals described here are those of 3.7.0; check the release
you run.
I help engineering teams untangle distributed systems, platform, and delivery problems. If that sounds like your week, email me at alex@bularca.me.