# Anatomy of a Kafka Lock Convoy > How eight single-member consumers fetching 4,742 partitions each stalled a ten-broker Kafka 3.7 cluster one broker at a time, and why the metrics never noticed. Published 2026-09-11 · Canonical: https://alex.bularca.me/posts/anatomy-of-a-kafka-lock-convoy/ 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 `DelayedOperationPurgatory` as a single > `DelayedFetch` registered under every partition it covers. Every produce, follower fetch and > consumer fetch that completes on any of those partitions calls `checkAndComplete`, which takes > that one operation's `ReentrantLock` and 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 pipeline. Everything in this post happens inside the brokers and in the publisher immediately in front of them. The replication job is an ordinary consumer; its shape is what made it dangerous.

## 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-ratio` > plus `io-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's > `io-ratio + io-wait-ratio` collapsed 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.

What the thread dumps showed. Handler CPU went down while the broker was "busy", because parked threads burn nothing. The controller-plane handler pool on the same JVM was idle throughout; only the data plane was involved.

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

One object, one lock, hundreds of doors into it. This is a correct and ordinary Kafka data structure. It assumes no single waiting operation is watched under hundreds of keys on one broker, and that its completion check is cheap. Both assumptions fail for a single-member consumer that fetches thousands of partitions while millions of messages behind.

Here is the reconstructed stack the parked handlers shared, abridged from the dumps: ```text "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.run ``` Thirty 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.

Why the outside saw a frozen broker while the inside looked fine. Time a request spends unread in a socket is invisible to every RequestMetrics timer. The publishers waited 30 to 240 seconds; the broker's own Produce p99 stayed under a few seconds.

### 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: ```scala // 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()) // blocks ``` [KAFKA-8334](https://issues.apache.org/jira/browse/KAFKA-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.

Leadership is the selector. When two repaired brokers rejoined and a leadership-only reassignment equalised leaders to about 980 per broker, Broker A fell to about 470 replicated partitions and released within the next two-minute scrape. That is a margin, not a fix: any restart, node loss or rebalance re-arms it.

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: ```csharp // 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: 1. **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. 2. **The real error is unobservable.** The token cancels the _continuation_, not the produce. The delivery report, and with it `Local: Message timed out` or 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. 3. **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. 4. **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.ms` has nothing to batch.

Two systems, one loop. The consumers made the brokers fragile; the producer converted fragility into failures, duplicates and extra load. Each side can break the loop independently.

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: ```csharp var pending = batch.Select(msg => { var tcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); try { producer.Produce(topicPartition, msg, report => tcs.TrySetResult(report)); // non-blocking } catch (ProduceException 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 PersistenceStatus ``` Two 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.

2.8× throughput at a fifth of the latency, within three minutes, with nothing else changed. Messages appended on the brokers tripled, so this is delivery, not retries. On the broker that had carried the early signature, follower-fetch local p75 went from 12 to 15 ms to 1 ms and its request queue from 500 to single digits in the same minute; Produce purgatory size collapsed on every broker.

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 | When | What | | -------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | 11 Jun | Cluster migrated from a 2.x release to 3.7.0. First broker-side lag incident two days later; two more in August, each "solved" by moving load off a slow node. | | 2 Sep 17:43 | Hardware migration concentrates leadership on two freshly moved brokers. One replication consumer closes and reopens (new client id), issuing a fresh full fetch across all 4,742 partitions. | | 2 Sep 17:46 | Publisher begins failing every batch at its 120 s budget. Onset. | | 3 – 7 Sep | 1.6 – 1.85 M failed batch executions per day. Broker logs: not one WARN or ERROR during any stall. Consumers 2 – 40 h behind. A broker leaves for node retirement on 4 Sep; the cluster runs on nine. | | 5 Sep 08:00 | The publisher's upstream queue is cleared during the incident. About 36 hours of events never reach Kafka and are later re-sent from the source database. No dead-letter path existed. | | 7 Sep 15:50 | Leadership of the hot topic moved off the stalled broker; failures drop 15× that second. The convoy moves to the next most-loaded broker for two hours, then subsides. Evening episodes on 8 and 9 Sep. | | 10 Sep 07:16 – 12:30 | Three multi-terabyte rebalances approved and stopped in turn; each landing of replicas on the most-loaded node triggers a short episode. | | 10 Sep 12:45 | Storage-controller failure on one node: two brokers crash, 14 partitions offline, 140 under min-ISR, `NOT_ENOUGH_REPLICAS` at 440/s. Eight brokers left. | | 10 Sep 13:18 | Broker A, now leading 35 % more partitions, blocks completely with no CPU, disk or reassignment pressure. | | 10 Sep 14:10 | Three thread dumps. The lock holder is named; partitions led per broker match the stall ranking. | | 10 Sep 15:31 – 15:49 | Repaired node returns; preferred-leader election, then a leadership-only reassignment equalises leaders. Broker A releases within two minutes. | | 10 Sep 16:48 | Last failed batch, after a 75-minute tail of failures at exactly the 240 s budget with no Kafka error: the amplifier alone. | | 10 Sep 17:41 | Eight replication consumers stopped. Three minutes later: 2.8× throughput, ⅕ latency, zero errors. First clean night. | | 11 Sep 08:06 | Rewritten producer deployed: non-blocking sends, per-message delivery reports, real error logged. Batches four times larger, twice as fast, zero errors. Wire still 1.06 msg/request until linger is raised. | ## 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.ms` are wrong for its width. Cap `fetch.max.bytes` and `max.partition.fetch.bytes` so 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.max` per 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_percentage` quota 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.threads` or `queued.max.requests` as 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.ms` below any application budget so librdkafka fails stragglers with a reportable reason before your own timer does. - Raise `linger.ms` deliberately 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 ```text # 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) > 300 ``` When 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.