Kafka listener shutdown hangs indefinitely: KafkaListener.StopAsync awaits BackgroundReceiveLoop with Timeout.InfiniteTimeSpan

[gardasar-code]

Summary

Since 6.6.0, an application using the Kafka transport never completes graceful
shutdown. IHost.StopAsync() blocks forever, so the process only terminates when it is
force-killed (e.g. Kubernetes SIGKILL after the termination grace period). Reproduced on
the current latest 6.19.0.

HostOptions.ShutdownTimeout does not help: the hang is below a StopAsync overload
that takes no CancellationToken, so the host cannot interrupt it.

Environment

• Wolverine + WolverineFx.Kafka: 6.19.0 (present since 6.6.0; 6.5.1 was the last release
  without it — see the version history below)
• Confluent.Kafka: 2.15.0
• .NET 10
• OS: macOS (Apple Silicon) and Linux containers

Expected

IHost.StopAsync() / host disposal completes within a bounded time when the Kafka
transport has active listeners.

Actual

IHost.StopAsync() never returns. A hang dump shows the shutdown blocked here (async
call chain, leaf first):

Wolverine.Kafka.Internals.KafkaListener.StopAsync            <-- awaiting, never completes
Wolverine.Transports.ListeningAgent.StopAndDrainCoreAsync
Wolverine.Transports.ListeningAgent.StopAndDrainAsync
Wolverine.Configuration.EndpointCollection.DrainAsync        (Task.WhenAll over endpoints)
Wolverine.Runtime.WolverineRuntime.StopAsync
Microsoft.Extensions.Hosting.Internal.Host.StopAsync

At the same time the listener's receive loop task has not completed. In the hang dump its
state machine appears as a running root (BackgroundReceiveLoop.runAsync, no pending
awaiter), i.e. it is synchronously parked inside the blocking IConsumer.Consume(token)
call — the first statement of consumeOnceAsync, before any await:

Wolverine.Transports.BackgroundReceiveLoop.runAsync
  -> KafkaListener.consumeOnceAsync
    -> IConsumer<string, byte[]>.Consume(CancellationToken)   (has not returned)

Root cause

At shutdown, KafkaListener.StopAsync cancels the token and then awaits the receive loop
with an infinite timeout (6.15.0+ form, via the shared BackgroundReceiveLoop):

public async ValueTask StopAsync()
{
    await _cancellation.CancelAsync();
    await _loop.StopAsync(Timeout.InfiniteTimeSpan);   // <-- unbounded
}

BackgroundReceiveLoop.StopAsync(TimeSpan timeout) waits on the loop task via
_task.WaitAsync(timeout) and already handles a finite budget gracefully (it catches
TimeoutException and logs "did not drain within {Timeout}"). But it is called with
Timeout.InfiniteTimeSpan, so it waits forever whenever the loop task does not complete.

The loop task does not complete because the synchronous _consumer.Consume(token) call at
the top of consumeOnceAsync has not returned and has not observed cancellation in this
scenario, so the while (!IsCancellationRequested) guard is never re-evaluated.

Version history (decompiled)

• 6.5.1 — fire-and-forget; shutdown never blocked:

  public async ValueTask StopAsync()
  {
      await _cancellation.CancelAsync();   // returns immediately; loop torn down later
  }

• 6.6.0 – 6.14.x — introduces an unbounded await of the loop task:

  await _cancellation.CancelAsync();
  await _runner;                           // unbounded

• 6.15.0 – 6.19.0 (latest) — same unbounded semantics, refactored onto the shared
  BackgroundReceiveLoop:

  await _cancellation.CancelAsync();
  await _loop.StopAsync(Timeout.InfiniteTimeSpan);   // unbounded

Reproduction

Observed with an ASP.NET host hosted under WebApplicationFactory<Program> in an
integration test. The host configures the Kafka transport with listeners declared as
opts.ListenToKafkaTopic(topic).ProcessInline().UseDurableInbox(), against a real Kafka
broker (Testcontainers). Sequence:

1. Start the host (so the Kafka listener / receive loop is running).
2. Dispose the host — WebApplicationFactory.DisposeAsync(), which invokes
   IHost.StopAsync().

IHost.StopAsync() never returns; the process only ends when the test runner's hang-dump
timeout aborts it. The hang dump shows the async chain in "Actual" above, rooted at
WebApplicationFactory<Program>.DisposeAsync -> Host.StopAsync -> WolverineRuntime.StopAsync -> EndpointCollection.DrainAsync -> KafkaListener.StopAsync (leaf).

A direct host.StopAsync() and a SIGTERM-driven graceful shutdown reach the same
Host.StopAsync path, so they are expected to hang identically.

Suggested fix

Pass a finite, bounded drain timeout to _loop.StopAsync(...) instead of
Timeout.InfiniteTimeSpan (ideally configurable, e.g. via DurabilitySettings or a
listener/endpoint option). BackgroundReceiveLoop.StopAsync already swallows the
TimeoutException and logs, so a finite value gives clean, bounded teardown.

Additionally (or alternatively), to preserve the "don't close the consumer while the loop
still uses it" guarantee without an unbounded wait: after the drain budget elapses, close/
dispose the IConsumer to force the blocking Consume(token) to unwind, then complete
teardown.