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

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

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())   // blocks

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.

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:

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.

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

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

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