Repository navigation
GH-4330: keep the blocking Kafka consume loop off the thread pool - #4354
Merged
Merged
Conversation
IConsumer.Consume(CancellationToken) blocks synchronously, and the loop calling it runs on Task.Run -- so every Kafka listener permanently occupies a thread-pool worker for its whole lifetime. Multiply by topics and ListenerCount and it eats the same pool the handler pipeline and every other transport's continuations share, which is classic starvation under load. BackgroundReceiveLoop gains an opt-in UsesBlockingIteration that starts the loop with TaskCreationOptions.LongRunning (a dedicated thread) and Kafka sets it. Deliberately opt-in: the SQS, Redis and database-queue loops are genuinely async and release their thread at every await, so a dedicated thread there would waste one. KafkaOffsetCommitter allocated a TopicPartition -- a class -- on every Track (per consumed record) and every Complete (per handled record), purely to key the watermark lookup. The (topic, partition) set a listener sees is small and fixed by its assignment, so they are interned now. NOT done, deliberately: sharding the committer's single _lock per partition, and widening the batched-drain gate beyond Durable. The lock also guards the flush loop's iteration over all watermarks, and the drain's benefit outside Durable is amortized receiver bookkeeping rather than the inbox-insert win GH-3490 actually measured. Both are real candidates; neither should land on reasoning alone. Closes #4330 Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
jeremydmiller
added a commit
that referenced
this pull request
Sep 13, 2026
…rever (#4429) StopAsync bounded the receive-loop drain (GH-3434) and then did two unbounded things, so the bound bought nothing. _committer.Flush() is a synchronous _consumer.Commit(), and _consumer.Close() is a synchronous P/Invoke into rd_kafka_consumer_close, which waits on an infinite queue pop for the cgroup to finish revoke/commit/leave. librdkafka documents that wait as "roughly limited to session.timeout.ms", but against an unreachable broker or coordinator it never returns (librdkafka#4519, confluent-kafka-dotnet#2013). A reporter's host was still wedged 20+ minutes later, past both DrainTimeout and HostOptions.ShutdownTimeout, with IHost.StopAsync never completing. The try/catch around Close() could never help: it catches exceptions, not a blocked native call. Neither can a token -- IListener.StopAsync has none to thread, and a managed token cannot interrupt a native P/Invoke regardless. Bounding the wait is the only lever, so take the trade the receivers and the GCP Pub/Sub listener (#4071) already take: stop within the drain budget and leave the rest unsettled for redelivery. Both blocking steps now run under runTeardownWithinBudgetAsync, on a DEDICATED background thread rather than the pool. If the step wedges, that thread is never coming back, and abandoning a pool thread would permanently consume a worker shared by the handler pipeline and every other transport -- the same amplifier GH-4354 removed from the consume loop, reintroduced at shutdown. An abandoned teardown also suppresses the later consumer Dispose. Confluent's ReleaseHandle passes RD_KAFKA_DESTROY_F_NO_CONSUMER_CLOSE so disposing does not block, but the abandoned thread still owns that handle, and destroying it underneath a live native call would trade a hang for a crash. Deliberately NOT done here: threading a CancellationToken through DrainAsync/StopAndDrainAsync/IListener.StopAsync, which the issue offers as an option. That is a breaking signature change across every transport's listener, and it would not fix this anyway, because the blocked call is native. Three tests, no broker -- a substituted IConsumer is the only way to reproduce "never returns" deterministically. Each bounds its own wait, so a regression fails the test rather than hanging the suite: the close is still attempted and StopAsync returns; an abandoned close does not get its consumer disposed; and a healthy consumer is still closed AND disposed, so the ordinary path is unchanged. Claude-Session: https://claude.ai/code/session_01VDUrBeB4tTnKj4AExCS1nj Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Closes #4330. From the 2026-09-05 hot-path performance audit (GH-4316 wave).
A blocking loop on a pool thread.
IConsumer.Consume(CancellationToken)blocks synchronously, andBackgroundReceiveLoopstarts its iteration withTask.Run— so every Kafka listener permanently occupies a thread-pool worker for its entire lifetime. With several topics, orListenerCount > 1, that is several workers removed from the same pool the handler pipeline and every other transport's continuations run on: classic starvation exactly under the load Kafka is chosen for.BackgroundReceiveLoopgains an opt-inUsesBlockingIterationthat starts the loop withTaskCreationOptions.LongRunning(a dedicated thread), and Kafka sets it. Deliberately opt-in rather than the default: the SQS, Redis, Pulsar and database-queue loops are genuinely async and release their thread at every await, so a dedicated thread there would waste one.KafkaOffsetCommitterallocations.Trackruns per consumed record andCompleteper handled record, and both allocated aTopicPartition(a class, with its own hash/equals) purely to key the watermark lookup. The(topic, partition)set a listener sees is small and fixed by its partition assignment, so they're interned in aConcurrentDictionarynow.What I deliberately did not do
Two items from the issue are not in this PR, and I want to be explicit rather than quietly drop them:
_lockper partition. That lock also guards the flush loop's iteration over every watermark, so per-partition locking is a real concurrency refactor, not a mechanical change.Durable. The zero-timeout drain is indeed free, but the benefit outside Durable is amortized receiver bookkeeping — not the durable-inbox-insert win Kafka performance deep dive: measuring and reducing Wolverine overhead vs native Confluent.Kafka #3490 actually measured. The upside is speculative and the settlement interaction with NativeAck's lease tracking is subtle.Both are legitimate candidates. Neither should land on reasoning alone — and today's #4353 regression, where exactly that kind of reasoning removed a load-bearing guard, is why.
Testing
dotnet build wolverine.slnx -c Release -f net9.0clean (the pinned CI gate)BackgroundReceiveLoop's own lifecycle tests on the unchanged default path)CIKafkalane is the real gate for the listener change and is queued for tonight's local run🤖 Generated with Claude Code