Bound the Kafka listener shutdown drain instead of awaiting forever (GH-3434)

[jeremydmiller]

Fixes #3434.

Problem

Since 6.6.0, an app using the Kafka transport never completes graceful shutdown — IHost.StopAsync() blocks forever, so the process only terminates on a force-kill (e.g. Kubernetes SIGKILL after the grace period). Reproduced through 6.19.0.

KafkaListener.StopAsync (and Dispose, and their KafkaTopicGroupListener twins) drained the shared BackgroundReceiveLoop with Timeout.InfiniteTimeSpan. When the loop was parked inside a blocking IConsumer.Consume(token) that hadn't observed cancellation, the drain — and therefore host shutdown — never returned. HostOptions.ShutdownTimeout couldn't help because the blocking overload takes no CancellationToken.

Fix

Drain with the configurable, bounded DurabilitySettings.DrainTimeout (default 30s) instead of Timeout.InfiniteTimeSpan, in both StopAsync and Dispose. This matches the convention the SQS (_drainTimeout) and RDBMS (_settings.DrainTimeout) listeners already follow — Kafka was the lone Infinite outlier.

BackgroundReceiveLoop.StopAsync already catches the TimeoutException and logs, so this gives clean bounded teardown. On a timed-out drain, the existing _consumer.Close() that runs next forces a wedged Consume to unwind (the issue's alternative suggestion, for free). The timeout is plumbed through all four build sites, so per-tenant (broker-per-tenant) listeners get bounded shutdown as well. Configurable via opts.Durability.DrainTimeout.

Tests

Adds a BackgroundReceiveLoop regression test: an iteration that blocks while ignoring its cancellation token (the wedged-Consume shape) still lets StopAsync(finiteBudget) return within budget instead of hanging — an infinite await would never return.

Full local Wolverine.Kafka.Tests run against the docker-compose broker: 230 passed, 2 skipped, 1 pre-existing [Flaky] failure (batch_processing_with_kafka.end_to_end, a batch-composition timing assertion) that fails identically on pristine main and is unrelated to the shutdown path.

🤖 Generated with Claude Code