Skip to content

Keep a blocking receive loop on its dedicated thread past the first await (GH-4330 follow-up) - #5096

Merged
jeremydmiller merged 1 commit into
JasperFx:mainfrom
lahma:kafka-dedicated-consume-thread
Oct 20, 2026
Merged

jeremydmiller merged 1 commit into
JasperFx:mainfrom
lahma:kafka-dedicated-consume-thread

Conversation

@lahma

@lahma lahma commented Oct 20, 2026

Copy link
Copy Markdown
Contributor

Follow-up to GH-4330 (#4354), which set out to keep the Kafka listener's blocking IConsumer.Consume(CancellationToken) off the thread pool. As shipped, it only does so for the loop's first pass.

The problem

BackgroundReceiveLoop.Start() runs a loop marked UsesBlockingIteration as Task.Factory.StartNew(runAsync, ..., TaskCreationOptions.LongRunning, TaskScheduler.Default).Unwrap(). But runAsync is async and awaits each iteration with ConfigureAwait(false). KafkaListener.consumeOnceAsync calls Consume(token) synchronously and then awaits _receiver.ReceivedAsync(...), which on a durable endpoint is a real inbox insert. The LongRunning thread therefore runs the loop only until the first await that does not complete synchronously. Then that thread exits, the loop continues on the pool thread that completed the await, and every later Consume(token) blocks a pool worker until a record arrives. Each Kafka listener holds a pool worker for its whole lifetime, which GH-4330 was meant to prevent. The idle wait and the failure backoff in the loop do the same.

KafkaTopicGroupListener (ListenToKafkaTopics) runs the same blocking Consume(token) loop and never set UsesBlockingIteration, so it was on the pool from the start.

What we measured: a 1 vCPU container on .NET 10 with 3-4 durable Kafka listeners per process. The pool starts with one worker per CPU and adds roughly one every 500 ms. When a burst of 2,886 records over 12 partitions arrived, every HTTP and gRPC call in the process queued. p99 was about 650 ms, even for an endpoint that touches no database, against 2-20 ms between bursts. Removing the CPU quota did not help (600-780 ms). DOTNET_ThreadPool_ForceMinWorkerThreads=16 brought it back to 2-21 ms, and a raised minimum is our workaround for now.

The change

  • BackgroundReceiveLoop: one pass of the loop (heartbeat, iteration, EntityMissing/status handling, then the idle wait or the failure backoff) moves unchanged into runOnceAsync(), which returns false once the loop has been cancelled.
  • runAsync() awaits that pass in a loop exactly as before. Loops without UsesBlockingIteration (SQS, Redis, the database queues) still start on Task.Run and behave as before.
  • A loop with UsesBlockingIteration starts runOnDedicatedThread() with LongRunning. That driver waits on each pass with GetAwaiter().GetResult() instead of awaiting it, so the loop is back on its own thread before every iteration, and Consume(token) always blocks that thread. Whatever a pass awaits still completes on the pool. The thread has no SynchronizationContext and nothing posts back to it, so the wait cannot deadlock.
  • Cancellation, the bounded StopAsync drain and DisposeAsync work as before: Consume(token) throws on cancellation, the pass returns false, and the driver exits. Exceptions go through the same log, backoff and status path, because both drivers run the same pass.
  • KafkaTopicGroupListener now sets UsesBlockingIteration = true, as KafkaListener does.

I chose this shape because it keeps a single copy of the loop logic and the existing LongRunning start, and leaves async loops alone. A single-threaded SynchronizationContext or TaskScheduler for the loop would also work. It would mean dropping the loop's ConfigureAwait(false) calls, though, and it would run the receiver's continuations on the consume thread too.

Tests

New CoreTests/Transports/background_receive_loop_blocking_iteration_4330.cs:

  • the_blocking_part_of_every_iteration_runs_on_one_dedicated_thread: the iteration records its calling thread and then awaits Task.Delay. Every recorded call must be on one non-pool thread. On main the first call runs on the LongRunning thread and calls 2-5 run on pool threads, so the test fails. With this change it passes.
  • failures_still_back_off_and_recover_on_the_dedicated_thread: the failure/backoff path keeps running, recovers, and stays off the pool. Fails on main, passes with this change.
  • dispose_stops_an_iteration_blocked_the_way_consume_blocks: the iteration blocks on the token and throws on cancellation, as Consume(token) does. DisposeAsync returns promptly and the status is Stopped.
  • a_loop_that_does_not_block_stays_on_the_pool: the default path is unchanged.

End to end, I ran a throwaway probe (not part of this PR) against a real broker and a Postgres inbox. It used one durable ListenToKafkaTopic and one durable ListenToKafkaTopics group, 60 records, and statistics.interval.ms=100. librdkafka serves the statistics callback from inside Consume() on the calling thread, so I counted the callbacks that arrived on a pool thread after every record had been handled:

single-topic listener topic group listener
main 27 of 27 on a pool thread 27 of 27 on a pool thread
this PR 0 of 28 (one dedicated thread) 0 of 27 (one dedicated thread)

Ran locally on Windows. Only the .NET 10 runtime is installed here, so tests ran on net10.0, and the receive-loop classes also ran on the net9.0 build rolled forward to it:

  • dotnet build wolverine.slnx -c Release -f net9.0: 0 errors. The only warnings were the existing NU1902 audit warnings for MessagePack 3.1.7.
  • The background_receive_loop*, receive_loop_entity_missing_4215 and receive_loop_health_default_3236 classes: 18/18 passed on net10.0 and net9.0. The new class passed 25 runs in a row.
  • Full CoreTests on net10.0: 3702 tests, 7 failed, none of them in receive-loop code. Four failed only because my shell had DOTNET_ENVIRONMENT=Development set, which turns on scope validation; they pass without it. The other three are event_model_scaffold_4832, which compares multi-line string literals with generated code and fails in my Windows CRLF checkout.
  • Wolverine.Kafka.Tests on net10.0 (excluding Category=Flaky), against the docker-compose.yml images (confluent-local:8.0.6, postgres:17): 327/327 passed.

…wait

JasperFxGH-4330 started a BackgroundReceiveLoop marked UsesBlockingIteration with
TaskCreationOptions.LongRunning, so that the Kafka consumer's blocking
Consume(CancellationToken) would not hold a thread-pool worker. The loop it
started is async and awaits each iteration with ConfigureAwait(false), so the
dedicated thread only ran it up to the first await that did not complete
synchronously: for a Kafka listener, the receiver's work on the first record,
which on a durable endpoint is the inbox insert. The thread then exited, the
loop resumed on the pool thread that completed that await, and every later
Consume(token) blocked a pool worker again, one per listener for as long as
it waited for a record.

A loop marked UsesBlockingIteration now runs a synchronous driver on the
LongRunning thread that waits on each pass instead of awaiting it, so the loop
is back on that thread before it calls the iteration again. The pass itself
(heartbeat, iteration, EntityMissing handling, idle wait, failure backoff) is
moved unchanged into one method that both drivers share, and loops without
the flag still start on Task.Run and await it exactly as before.

KafkaTopicGroupListener runs the same blocking Consume(token) loop and never
set UsesBlockingIteration, so it now does.
Copilot AI balanced review requested due to automatic review settings October 20, 2026 07:19

Copilot AI left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟢 Approval recommended

The focused refactor preserves existing iteration handling, adds regression coverage, and has no identified blocking issues.

No open issues 🎉

What changed in this PR

Fixes the GH-4330 follow-up by keeping blocking Kafka receive iterations on a dedicated thread after asynchronous waits.

Changes:

  • Shares iteration handling between async and dedicated-thread drivers.
  • Enables dedicated-thread execution for Kafka topic groups.
  • Adds regression tests for thread affinity, failure recovery, disposal, and the default async path.
File Description
src/​Wolverine/​Transports/​BackgroundReceiveLoop.cs Keeps blocking iterations on a dedicated thread while preserving shared lifecycle handling.
src/​Transports/​Kafka/​Wolverine.Kafka/​Internals/​KafkaTopicGroupListener.cs Enables the dedicated-thread driver for topic groups.
src/​Testing/​CoreTests/​Transports/​background_receive_loop_blocking_iteration_4330.cs Adds threading and lifecycle regression coverage.

Review effort: Balanced

@jeremydmiller
jeremydmiller merged commit 35bfa17 into JasperFx:main Oct 20, 2026
47 checks passed
@lahma
lahma deleted the kafka-dedicated-consume-thread branch October 20, 2026 14:59
This was referenced Oct 21, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants