Kafka listener shutdown: _consumer.Close() is unbounded after the bounded drain (follow-up to #3434)

[AndreiKopylov]

Describe the bug

KafkaListener.StopAsync() bounds the receive-loop drain (await _loop.StopAsync(_drainTimeout), GH-3434) and then calls _consumer.Close() with no timeout and no cancellation token:

public async ValueTask StopAsync()
{
    await _cancellation.CancelAsync();
    // ... the _consumer.Close() below forces that Consume to unwind - bounded teardown instead of an infinite await.
    await _loop.StopAsync(_drainTimeout);   // bounded by DurabilitySettings.DrainTimeout (30s)

    _committer.Flush();                     // synchronous _consumer.Commit(), unbounded
    try
    {
        _consumer.Close();                  // unbounded, no token  <-- wedges here
    }
    catch (Exception e)
    {
        _logger.LogDebug(e, "Error closing Kafka consumer on shutdown");
    }
}

Confluent.Kafka.Consumer.Close() -> SafeKafkaHandle.ConsumerClose() -> Librdkafka.consumer_close(handle) is a synchronous P/Invoke with no cancellation. librdkafka documents it as "This call will block until the consumer has revoked its assignment, ... committed offsets to broker, and left the consumer group. The maximum blocking time is roughly limited to session.timeout.ms", and implements the wait as rd_kafka_q_pop(rkq, RD_POLL_INFINITE, 0) (rdkafka.c:5281); the TERMINATE op it waits for is only produced once the cgroup finished revoke/commit/leave (rdkafka_cgrp.c:4061-4066). With a degraded broker/coordinator the call never returns, so IHost.StopAsync never completes.

The try/catch around it cannot help: it catches exceptions, not a blocked native call.

Stack captured from a full process dump (dotnet-dump collect, analysed with dumpasync and clrstack -all) of a wedged host - line numbers match 6.26.0:

WebApplicationFactory<Program>.DisposeAsync
  Microsoft.Extensions.Hosting.Internal.Host.StopAsync
    Host.ForeachService<IHostedService>
      Microsoft.Extensions.Hosting.BackgroundService.StopAsync
        Wolverine.Runtime.WolverineRuntime.StopAsync(CancellationToken)   WolverineRuntime.HostService.cs:386
          Wolverine.Configuration.EndpointCollection.DrainAsync()          EndpointCollection.cs:535-539
            Wolverine.Transports.ListeningAgent.StopAndDrainAsync()        ListeningAgent.cs:255
              ListeningAgent.StopAndDrainCoreAsync(bool)                   ListeningAgent.cs:290
                Wolverine.Kafka.Internals.KafkaListener.StopAsync()        KafkaListener.cs:258
                  Confluent.Kafka.Consumer<string,byte[]>.Close()
                    SafeKafkaHandle.ConsumerClose()
                      Librdkafka.consumer_close(IntPtr)
                        NativeMethods.rd_kafka_consumer_close(IntPtr, IntPtr)   <-- BLOCKED

The managed main thread is parked in TaskAwaiter.GetResult() on that chain, syncblk is empty (no monitor deadlock), and the JasperFx.Blocks.Block<...>.processAsync worker loops are idle. In our case the process was still wedged after 20+ minutes, long past the 30s DrainTimeout and the 30s HostOptions.ShutdownTimeout.

Why HostOptions.ShutdownTimeout cannot rescue this

1. The token is never propagated: WolverineRuntime.StopAsync(CancellationToken) calls _endpoints.DrainAsync() with no argument, EndpointCollection.DrainAsync() has no token in its signature, ListeningAgent.StopAndDrainAsync() has none, and IListener.StopAsync() has no CancellationToken overload at all.
2. Even though Host.StopAsync builds a linked CTS with CancelAfter(ShutdownTimeout), Host.ForeachService simply awaits the grouped service tasks - cancelling the token does not interrupt a service that ignores it (and a native P/Invoke cannot be interrupted by a managed token anyway).

So the shutdown timeout fires and nothing changes.

To Reproduce

Intermittent, not deterministic: 2 of 20 runs (~10%) in our harness, always during host disposal after all tests had passed.

1. A Wolverine service with several Kafka listeners - ours has three opts.ListenToKafkaTopic(...), each with .ExtendConsumerConfiguration(c => c.GroupId = ...) and .ProcessConcurrentlyByKey(PartitionSlots.Three), plus AutoProvision() and a durable PostgreSQL outbox/inbox.
2. Host it in an integration test with WebApplicationFactory<Program>, RunWolverineInSoloMode().
3. Run the suite, then dispose the factory in NUnit [OneTimeTearDown] (which also calls IMessageStore.Admin.ClearAllAsync() just before).
4. Occasionally StopAsync never returns and the process stays alive forever; the test run is green, only the exit is missing.

Expected behavior

Shutdown should finish within a bounded budget. "Stop within the drain timeout, leave the rest unacked for redelivery" is preferable to "wait forever" - that is the trade the receivers already make (BufferedReceiver.cs:246, DurableReceiver.cs:424, InlineReceiver.cs:64 all bound themselves with DrainTimeout), and the same trade you applied to GCP Pub/Sub in #4071.

Screenshots

N/A - stack traces above.

Desktop

• OS: Windows 11 (Linux not verified)
• .NET SDK 10.0.100, net10.0
• WolverineFx / WolverineFx.Kafka 6.26.0; the same code path is present in 6.36.0 (_consumer.Close() at KafkaListener.cs:265)
• Confluent.Kafka 2.15.0 (librdkafka 2.15.0)
• Kafka 4.3.1 KRaft, single broker, no SSL

Additional context

Suggested fix, in the spirit of #3436 and #4065/#4071:

1. Thread the budget down: DrainAsync(CancellationToken) -> StopAndDrainAsync(CancellationToken) -> IListener.StopAsync(CancellationToken), defaulting to DurabilitySettings.DrainTimeout.
2. Bound the close the same way the loop drain is bounded, e.g. await Task.Run(() => _consumer.Close()).WaitAsync(_drainTimeout) - log and continue teardown on timeout.
3. Or, once the budget is exhausted, fall back to Unsubscribe() + Dispose(): Confluent.Kafka's ReleaseHandle uses RD_KAFKA_DESTROY_F_NO_CONSUMER_CLOSE, which does not block. The documented cost is no offset commit and no LeaveGroup, i.e. a rebalance after session.timeout.ms.

Data point on #4354: after upgrading 6.26.0 -> 6.36.0 the hang stopped reproducing for us - 0 of 32 runs versus 2 of 20 on 6.26.0, identical tests and broker. The only change in KafkaListener.cs between those tags is #4354 (UsesBlockingIteration = true, moving the blocking Consume off the thread pool). With three listeners, 6.26.0 permanently occupied three thread-pool workers, which is a plausible amplifier: librdkafka invokes the managed rebalance_cb synchronously inside rd_kafka_consumer_close, and that callback needs a pool thread. That is evidence about the amplifier, not a fix - Close() is still unbounded in 6.36.0, so the wedge should remain reachable whenever the broker or coordinator is unreachable.

Related upstream issues: librdkafka #4519 (open, "rd_kafka_consumer_close/rd_kafka_destroy remain blocked indefinitely if the broker is unreachable"), confluent-kafka-dotnet #2013 (open, consumer hangs on dispose, reported at ~20% frequency), and #4116 where the Kafka teardown hypothesis was called "weaker, but not eliminated".