< Summary - AsyncResponse (Release / net8.0+net10.0 / unit+integration)

Information
Class: AsyncResponse.Transports.Kafka.QueuedKafkaMessageDispatcher
Assembly: AsyncResponse.Transports.Kafka
File(s): /_/src/Transports/AsyncResponse.Transports.Kafka/KafkaMessageDispatcher.cs
Line coverage
94%
Covered lines: 141
Uncovered lines: 9
Coverable lines: 150
Total lines: 1309
Line coverage: 94%
Branch coverage
100%
Covered branches: 12
Total branches: 12
Branch coverage: 100%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%11100%
get_PendingCount()100%11100%
get_RunningCount()100%11100%
get_CanAcceptMore()100%11100%
HandleAsync()100%2294.44%
DisposeAsync()100%22100%
RunWorkerAsync()100%4479.48%
ExecuteWithRetryAsync()100%44100%

File(s)

/_/src/Transports/AsyncResponse.Transports.Kafka/KafkaMessageDispatcher.cs

#LineLine coverage
 1using Microsoft.Extensions.Logging;
 2using System.Diagnostics;
 3using System.Globalization;
 4using System.Text;
 5using System.Threading.Channels;
 6
 7namespace AsyncResponse.Transports.Kafka;
 8
 9internal enum KafkaSubscriberRole
 10{
 11    Worker,
 12    ResponseIngress
 13}
 14
 15internal sealed record KafkaDelivery(
 16    string Topic,
 17    int Partition,
 18    long Offset,
 19    string Payload,
 20    string? CorrelationId,
 21    IReadOnlyList<KafkaTransportHeader> Headers);
 22
 23/// <summary>
 24/// Turns consumed Kafka messages into AsyncResponse handler invocations under one of the two ACK
 25/// modes. Kafka offsets cannot NACK a single message, so failed handlers are retried in-process
 26/// with bounded backoff; a message that exhausts its attempts is produced to the dead-letter topic
 27/// and its offset stored so the partition keeps moving.
 28/// </summary>
 29/// <summary>
 30/// A message exhausted its handling and its dead-letter publish failed for good, so it is
 31/// neither buried nor committable. Thrown out of the poll loop ON PURPOSE: Kafka commits a
 32/// partition <em>position</em>, not per-record acknowledgements, so merely leaving this message's
 33/// offset unstored (the previous behavior) protected nothing — the next successful settlement on
 34/// the same partition stored a higher offset, the auto-committer committed past the failed
 35/// message, and a restart skipped it with no dead-letter copy anywhere. Faulting the subscriber
 36/// instead stops the partition at the unresolved message: the consumer closes without ever
 37/// storing past it, the supervisor rebuilds it after its backoff, and the message is re-consumed
 38/// and its burial retried until the dead-letter topic is back. That is a loud, bounded-rate loop
 39/// (every restart logs this failure) and a stalled subscriber — the at-least-once outcome — rather
 40/// than a silent loss.
 41/// </summary>
 42internal sealed class KafkaDeadLetterPublishFailedException(string topic, int partition, long offset, Exception innerExc
 43    : Exception(
 44        $"Kafka message {topic}[{partition}]@{offset} could not be dead-lettered after exhausting its handling attempts.
 45        "Its offset is left unstored and the subscriber is restarted so no later settlement on the partition commits pas
 46        "fix the dead-letter topic to let the partition advance.",
 47        innerException)
 48{
 49    public string Topic { get; } = topic;
 50    public int Partition { get; } = partition;
 51    public long Offset { get; } = offset;
 52}
 53
 54internal abstract class KafkaMessageDispatcher : IAsyncDisposable
 55{
 56    private readonly Func<KafkaDelivery, CancellationToken, Task> _handler;
 57    private readonly KafkaSubscriberOptions _subscriberOptions;
 58    private readonly IKafkaConsumerClient _consumer;
 59    private readonly IKafkaProducerClient _producer;
 60    private readonly KafkaTransportTopicSchema _topics;
 61    private readonly string _topic;
 62    private readonly string _consumerGroup;
 63    private readonly KafkaSubscriberRole _role;
 64
 65    /// <summary>Runs the KafkaMessageDispatcher operation.</summary>
 66    protected KafkaMessageDispatcher(
 67        Func<KafkaDelivery, CancellationToken, Task> handler,
 68        IKafkaConsumerClient consumer,
 69        IKafkaProducerClient producer,
 70        KafkaAsyncResponseTransportOptions transportOptions,
 71        KafkaSubscriberOptions subscriberOptions,
 72        ILogger logger,
 73        string topic,
 74        string consumerGroup,
 75        KafkaSubscriberRole role)
 76    {
 77        _handler = handler;
 78        _consumer = consumer;
 79        _producer = producer;
 80        TransportOptions = transportOptions;
 81        _subscriberOptions = subscriberOptions;
 82        _topics = new KafkaTransportTopicSchema(transportOptions);
 83        Logger = logger;
 84        _topic = topic;
 85        _consumerGroup = consumerGroup;
 86        _role = role;
 87    }
 88
 89    protected KafkaAsyncResponseTransportOptions TransportOptions { get; }
 90    protected ILogger Logger { get; }
 91    protected IKafkaConsumerClient Consumer => _consumer;
 92
 93    protected int MaxDeliveryAttempts => _subscriberOptions.MaxDeliveryAttempts;
 94
 95    /// <summary>Creates the configured dispatcher.</summary>
 96    public static KafkaMessageDispatcher Create(
 97        Func<KafkaDelivery, CancellationToken, Task> handler,
 98        IKafkaConsumerClient consumer,
 99        IKafkaProducerClient producer,
 100        KafkaAsyncResponseTransportOptions transportOptions,
 101        KafkaSubscriberOptions subscriberOptions,
 102        ILogger logger,
 103        string topic,
 104        string consumerGroup,
 105        KafkaSubscriberRole role)
 106    {
 107        ValidateOptions(transportOptions, subscriberOptions, role);
 108
 109        if (subscriberOptions.AckMode is KafkaAckMode.AckAfterEnqueue)
 110        {
 111            return new QueuedKafkaMessageDispatcher(
 112                handler,
 113                consumer,
 114                producer,
 115                transportOptions,
 116                subscriberOptions,
 117                logger,
 118                topic,
 119                consumerGroup,
 120                role);
 121        }
 122
 123        return new AwaitingKafkaMessageDispatcher(
 124            handler,
 125            consumer,
 126            producer,
 127            transportOptions,
 128            subscriberOptions,
 129            logger,
 130            topic,
 131            consumerGroup,
 132            role);
 133    }
 134
 135    /// <summary>Validates the supplied options.</summary>
 136    public static void ValidateOptions(
 137        KafkaAsyncResponseTransportOptions transportOptions,
 138        KafkaSubscriberOptions subscriberOptions,
 139        KafkaSubscriberRole role)
 140    {
 141        KafkaTransportOptionsValidator.ValidateCommon(transportOptions);
 142
 143        var optionPath = role is KafkaSubscriberRole.Worker
 144            ? $"{nameof(KafkaAsyncResponseTransportOptions)}.{nameof(KafkaAsyncResponseTransportOptions.WorkerSubscriber
 145            : $"{nameof(KafkaAsyncResponseTransportOptions)}.{nameof(KafkaAsyncResponseTransportOptions.ResponseSubscrib
 146
 147        // PollTimeout and BackpressurePollDelay both go to Consume(TimeSpan), which librdkafka
 148        // takes as 32-bit milliseconds; the handler-retry delays arm in-process Task.Delay timers.
 149        KafkaTransportOptionsValidator.EnsureIntMilliseconds(subscriberOptions.PollTimeout, optionPath, nameof(KafkaSubs
 150        KafkaTransportOptionsValidator.EnsureIntMilliseconds(subscriberOptions.BackpressurePollDelay, optionPath, nameof
 151        if (subscriberOptions.MaxDeliveryAttempts < 0)
 152            throw new InvalidOperationException($"{optionPath}.{nameof(KafkaSubscriberOptions.MaxDeliveryAttempts)} cann
 153        AsyncResponseChannelOptions.EnsureTimerBacked(subscriberOptions.HandlerRetryBaseDelay, optionPath, nameof(KafkaS
 154        AsyncResponseChannelOptions.EnsureTimerBacked(subscriberOptions.HandlerRetryMaxDelay, optionPath, nameof(KafkaSu
 155        if (subscriberOptions.HandlerRetryBaseDelay > subscriberOptions.HandlerRetryMaxDelay)
 156        {
 157            throw new InvalidOperationException(
 158                $"{optionPath}.{nameof(KafkaSubscriberOptions.HandlerRetryBaseDelay)} cannot exceed " +
 159                $"{optionPath}.{nameof(KafkaSubscriberOptions.HandlerRetryMaxDelay)}.");
 160        }
 161
 162        KafkaTransportOptionsValidator.EnsureMaxPollInterval(subscriberOptions.MaxPollInterval, optionPath, nameof(Kafka
 163        AsyncResponseChannelOptions.EnsureTimerBackedAllowZero(subscriberOptions.DetachHandlerAfter, optionPath, nameof(
 164        AsyncResponseChannelOptions.EnsureTimerBackedAllowZero(subscriberOptions.FaultDrainTimeout, optionPath, nameof(K
 165
 166        switch (subscriberOptions.AckMode)
 167        {
 168            case KafkaAckMode.AckAfterHandlerCompletes:
 169                // The poll thread's longest gap in this mode is one inline handler wait plus one
 170                // poll; a gap reaching max.poll.interval.ms gets the consumer evicted from its
 171                // group and its partitions redelivered elsewhere. Half the interval is the margin.
 172                // Handler execution time and the in-process retry ladder no longer count: past
 173                // DetachHandlerAfter the handler runs detached while the poll thread keeps polling
 174                // (the earlier rule bounded the retry DELAYS for that reason, and left real handler
 175                // time — a flow step awaiting a remote response — to overrun the interval anyway).
 176                var pollGapMs = subscriberOptions.DetachHandlerAfter.TotalMilliseconds + subscriberOptions.PollTimeout.T
 177                if (pollGapMs * 2 > subscriberOptions.MaxPollInterval.TotalMilliseconds)
 178                {
 179                    throw new InvalidOperationException(
 180                        $"{optionPath}: {nameof(KafkaSubscriberOptions.DetachHandlerAfter)} ({subscriberOptions.DetachHa
 181                        $"{nameof(KafkaSubscriberOptions.PollTimeout)} ({subscriberOptions.PollTimeout}) must fit within
 182                        $"{nameof(KafkaSubscriberOptions.MaxPollInterval)} ({subscriberOptions.MaxPollInterval}) — that 
 183                        "longest gap, and a gap reaching max.poll.interval.ms gets the consumer evicted from its group. 
 184                        $"{nameof(KafkaSubscriberOptions.DetachHandlerAfter)} or raise {nameof(KafkaSubscriberOptions.Ma
 185                }
 186
 187                return;
 188
 189            case KafkaAckMode.AckAfterEnqueue:
 190                if (subscriberOptions.BackgroundWorkerCount <= 0)
 191                {
 192                    throw new InvalidOperationException(
 193                        $"{optionPath}.{nameof(KafkaSubscriberOptions.BackgroundWorkerCount)} must be explicitly configu
 194                        $"when {nameof(KafkaSubscriberOptions.AckMode)} is {nameof(KafkaAckMode.AckAfterEnqueue)}.");
 195                }
 196
 197                if (subscriberOptions.BackgroundQueueCapacity <= 0)
 198                {
 199                    throw new InvalidOperationException(
 200                        $"{optionPath}.{nameof(KafkaSubscriberOptions.BackgroundQueueCapacity)} must be explicitly confi
 201                        $"when {nameof(KafkaSubscriberOptions.AckMode)} is {nameof(KafkaAckMode.AckAfterEnqueue)}.");
 202                }
 203
 204                AsyncResponseChannelOptions.EnsureTimerBacked(subscriberOptions.BackgroundDrainTimeout, optionPath, name
 205
 206                // Kafka subscribers spend only the background drain at shutdown; the poll loop
 207                // stops with the host token and the consumer close is not separately bounded.
 208                ShutdownBudgetValidator.Validate(
 209                    "Kafka",
 210                    $"{nameof(KafkaAsyncResponseTransportOptions)}.{nameof(KafkaAsyncResponseTransportOptions.HostShutdo
 211                    transportOptions.HostShutdownTimeout,
 212                    ($"{optionPath}.{nameof(KafkaSubscriberOptions.BackgroundDrainTimeout)}", subscriberOptions.Backgrou
 213
 214                return;
 215
 216            default:
 217                throw new InvalidOperationException(
 218                    $"{optionPath}.{nameof(KafkaSubscriberOptions.AckMode)} has unsupported value '{subscriberOptions.Ac
 219        }
 220    }
 221
 222    /// <summary>Handles the delivered message through to settlement, offset store included.</summary>
 223    public abstract Task HandleAsync(KafkaDelivery delivery, CancellationToken subscriberCancellationToken);
 224
 225    /// <summary>
 226    /// The poll thread's entry point for a consumed message. Returns once the message is settled
 227    /// (offset stored, or dead-lettered and stored) or — for the awaiting dispatcher — once its
 228    /// handler has been detached to run on while polling continues. Throws when the message cannot
 229    /// be settled (a permanently failing burial, cancellation), which faults the poll loop so the
 230    /// subscriber is rebuilt without ever committing past the message.
 231    /// </summary>
 232    public virtual void Accept(KafkaDelivery delivery, CancellationToken subscriberCancellationToken)
 233        => HandleAsync(delivery, subscriberCancellationToken).GetAwaiter().GetResult();
 234
 235    /// <summary>
 236    /// The poll thread's entry point for a consumed message that could not be turned into a
 237    /// delivery (<see cref="DiscardUnprocessableAsync"/> describes the settlement). Default:
 238    /// settled at once — the queued dispatcher stores every offset at enqueue, in consumption
 239    /// order, so nothing earlier on the partition is still unresolved. The awaiting dispatcher
 240    /// overrides it to hold the message behind a detached handler of the same partition: its
 241    /// offset must not be stored — and so committed — ahead of a message consumed before it that
 242    /// is still being handled.
 243    /// </summary>
 244    public virtual void AcceptUnprocessable(KafkaIncomingMessage message, Exception failure, CancellationToken subscribe
 245        => DiscardUnprocessableAsync(message, failure, subscriberCancellationToken).GetAwaiter().GetResult();
 246
 247    /// <summary>
 248    /// Poll-thread tick: settles detached handlers that have finished — offset stored, partition
 249    /// resumed, the next held message started. Throws when one of them failed for good (the poll
 250    /// loop faults, exactly as an inline failure would).
 251    /// </summary>
 252    public virtual void SettleCompleted()
 253    {
 254    }
 255
 256    /// <summary>
 257    /// The poll loop FAILED (as opposed to a stop) and the consumer is about to be closed and
 258    /// rebuilt by the supervisor. Default: the graceful drain. The awaiting dispatcher overrides
 259    /// it with a bounded wait (<see cref="KafkaSubscriberOptions.FaultDrainTimeout"/>) so the
 260    /// reconnect is not parked behind an unrelated long handler.
 261    /// </summary>
 262    public virtual ValueTask TeardownAfterFaultAsync() => DisposeAsync();
 263
 264    /// <summary>
 265    /// Whether detached handlers are in flight. The poll loop then polls in
 266    /// <see cref="KafkaSubscriberOptions.BackpressurePollDelay"/> slices so a completion is settled
 267    /// promptly instead of after a full <see cref="KafkaSubscriberOptions.PollTimeout"/>.
 268    /// </summary>
 269    public virtual bool HasDetachedWork => false;
 270
 271    /// <summary>
 272    /// Whether the dispatcher can accept more deliveries right now. Awaiting dispatchers always can
 273    /// (a partition with a detached handler is paused, so nothing arrives for it); the queued
 274    /// dispatcher returns <c>false</c> while its bounded queue is saturated so the subscriber
 275    /// pauses partition fetching instead of buffering an unbounded backlog in-process.
 276    /// </summary>
 277    public virtual bool CanAcceptMore => true;
 278
 279    /// <summary>Releases resources held by this instance.</summary>
 280    public virtual ValueTask DisposeAsync() => ValueTask.CompletedTask;
 281
 282    /// <summary>Runs the ExecuteHandlerAsync operation.</summary>
 283    protected async Task ExecuteHandlerAsync(
 284        KafkaDelivery delivery,
 285        int attempt,
 286        CancellationToken cancellationToken,
 287        bool logFailures = true)
 288    {
 289        using var activity = AsyncResponseDiagnostics.StartActivity(
 290            "asyncresponse.kafka.receive",
 291            ActivityKind.Consumer,
 292            delivery.CorrelationId);
 293        activity?.SetTag("asyncresponse.transport", "kafka");
 294        activity?.SetTag("asyncresponse.kafka.role", _role.ToString());
 295        activity?.SetTag("asyncresponse.kafka.ack_mode", _subscriberOptions.AckMode.ToString());
 296        activity?.SetTag("asyncresponse.kafka.delivery_attempt", attempt);
 297        activity?.SetTag("messaging.system", "kafka");
 298        activity?.SetTag("messaging.destination.name", delivery.Topic);
 299        activity?.SetTag("messaging.kafka.consumer.group", _consumerGroup);
 300        activity?.SetTag("messaging.kafka.destination.partition", delivery.Partition);
 301        activity?.SetTag("messaging.kafka.message.offset", delivery.Offset);
 302
 303        try
 304        {
 305            await _handler(delivery, cancellationToken).ConfigureAwait(false);
 306        }
 307        catch (Exception ex)
 308        {
 309            if (logFailures)
 310            {
 311                Logger.LogError(
 312                    ex,
 313                    "Kafka message handling failed for {Topic}[{Partition}]@{Offset} (attempt {Attempt}).",
 314                    delivery.Topic,
 315                    delivery.Partition,
 316                    delivery.Offset,
 317                    attempt);
 318            }
 319
 320            AsyncResponseDiagnostics.SetError(activity, ex);
 321            throw;
 322        }
 323    }
 324
 325    /// <summary>
 326    /// Marks the delivered message resolved by storing its next offset; the consumer's
 327    /// auto-committer flushes stored offsets on the configured interval.
 328    /// </summary>
 329    protected void StoreOffset(KafkaDelivery delivery)
 330        => _consumer.StoreOffset(delivery.Topic, delivery.Partition, delivery.Offset);
 331
 332    /// <summary>
 333    /// Stores the offset after the message is settled (handler success or dead-letter publish),
 334    /// swallowing failures: StoreOffset throws when a rebalance revoked the partition — routine in
 335    /// a consumer group — and the only consequence is redelivery after the rebalance. A thrown
 336    /// settlement must never be misread as a handler failure that re-runs (or dead-letters)
 337    /// already-settled work.
 338    /// </summary>
 339    protected void StoreOffsetAfterSettlement(KafkaDelivery delivery)
 340        => StoreOffsetAfterSettlement(delivery.Topic, delivery.Partition, delivery.Offset);
 341
 342    /// <summary>
 343    /// Coordinate overload, for settlement paths that never built a <see cref="KafkaDelivery"/> —
 344    /// the unprocessable-message discard runs before projection succeeds.
 345    /// </summary>
 346    protected void StoreOffsetAfterSettlement(string topic, int partition, long offset)
 347    {
 348        try
 349        {
 350            _consumer.StoreOffset(topic, partition, offset);
 351        }
 352        catch (Exception ex)
 353        {
 354            Logger.LogError(
 355                ex,
 356                "Failed to store offset for Kafka message {Topic}[{Partition}]@{Offset} after settlement; it will be red
 357                topic,
 358                partition,
 359                offset);
 360        }
 361    }
 362
 363    /// <summary>Runs the ReachedDeliveryAttempts operation.</summary>
 364    protected bool ReachedDeliveryAttempts(int attempt)
 365        => MaxDeliveryAttempts > 0 && attempt >= MaxDeliveryAttempts;
 366
 367    /// <summary>Computes the delay before the next in-process handler retry.</summary>
 368    protected TimeSpan RetryBackoff(int completedAttempts)
 369        => AsyncResponseRetry.Backoff(
 370            completedAttempts,
 371            _subscriberOptions.HandlerRetryBaseDelay,
 372            _subscriberOptions.HandlerRetryMaxDelay);
 373
 374    /// <summary>
 375    /// Produces the failing message to the dead-letter topic (when enabled), preserving the
 376    /// original payload and headers and attaching failure-detail headers.
 377    /// </summary>
 378    protected async Task DeadLetterAsync(
 379        KafkaDelivery delivery,
 380        Exception exception,
 381        string reason,
 382        int attempts,
 383        CancellationToken cancellationToken)
 384        => await DeadLetterCoreAsync(
 385            delivery.Topic,
 386            delivery.Partition,
 387            delivery.Offset,
 388            Encoding.UTF8.GetBytes(delivery.Payload),
 389            delivery.Headers,
 390            delivery.CorrelationId,
 391            exception,
 392            reason,
 393            attempts,
 394            cancellationToken).ConfigureAwait(false);
 395
 396    /// <summary>
 397    /// Dead-letters (when enabled) and stores the offset of a message that could not be turned into
 398    /// a delivery — for example a foreign message with an empty payload. Without this, such a
 399    /// message would fail before <see cref="HandleAsync"/> runs on every subscriber restart and its
 400    /// partition would never advance.
 401    /// </summary>
 402    public async Task DiscardUnprocessableAsync(
 403        KafkaIncomingMessage message,
 404        Exception failure,
 405        CancellationToken cancellationToken)
 406    {
 407        Logger.LogError(
 408            failure,
 409            "Kafka message {Topic}[{Partition}]@{Offset} could not be parsed into a delivery; dead-lettering and committ
 410            message.Topic,
 411            message.Partition,
 412            message.Offset);
 413
 414        // Settlement ignores the stopping token, as every sibling settlement path does: a shutdown
 415        // landing between the dead-letter publish and the offset store would abort the publish
 416        // mid-flight and leave the poison message neither buried nor committed. A burial that
 417        // fails for good FAULTS the poll loop (see KafkaDeadLetterPublishFailedException): an
 418        // earlier round swallowed it and left the offset unstored, which looked safe but was not —
 419        // the next settlement on the same partition committed past this message. The restart loop
 420        // it replaces is bounded by the supervisor's backoff and is the at-least-once outcome.
 421        try
 422        {
 423            await DeadLetterCoreAsync(
 424                message.Topic,
 425                message.Partition,
 426                message.Offset,
 427                message.Payload ?? [],
 428                message.Headers,
 429                KafkaCorrelationIdExtractor.TryReadHeader(message.Headers, TransportOptions.CorrelationIdHeader),
 430                failure,
 431                "unprocessable_message",
 432                attempts: 0,
 433                CancellationToken.None).ConfigureAwait(false);
 434        }
 435        catch (Exception deadLetterException)
 436        {
 437            Logger.LogError(
 438                deadLetterException,
 439                "Failed to dead-letter unprocessable Kafka message {Topic}[{Partition}]@{Offset}; its offset is left uns
 440                message.Topic,
 441                message.Partition,
 442                message.Offset);
 443            throw new KafkaDeadLetterPublishFailedException(message.Topic, message.Partition, message.Offset, deadLetter
 444        }
 445
 446        // Guarded like every other settlement: a rebalance revoking this partition makes
 447        // StoreOffset throw, and here that throw originates INSIDE the poll loop's catch arm, so
 448        // nothing could catch it — it faulted the poll loop after the message was already produced
 449        // to the dead-letter topic, and the restart dead-lettered it a second time.
 450        StoreOffsetAfterSettlement(message.Topic, message.Partition, message.Offset);
 451    }
 452
 453    /// <summary>
 454    /// Longest <c>exceptionType</c> / <c>exceptionMessage</c> dead-letter header value, in UTF-16
 455    /// code units.
 456    /// </summary>
 457    internal const int MaxDeadLetterHeaderLength = 4096;
 458
 459    private async Task DeadLetterCoreAsync(
 460        string sourceTopic,
 461        int partition,
 462        long offset,
 463        byte[] payload,
 464        IReadOnlyList<KafkaTransportHeader> originalHeaders,
 465        string? correlationId,
 466        Exception exception,
 467        string reason,
 468        int attempts,
 469        CancellationToken cancellationToken)
 470    {
 471        if (!TransportOptions.DeadLetterEnabled)
 472            return;
 473
 474        var headers = new List<KafkaTransportHeader>(originalHeaders.Count + 10);
 475        headers.AddRange(originalHeaders);
 476        headers.Add(KafkaTransportHeader.Utf8("sourceTopic", sourceTopic));
 477        headers.Add(KafkaTransportHeader.Utf8("sourcePartition", partition.ToString(CultureInfo.InvariantCulture)));
 478        headers.Add(KafkaTransportHeader.Utf8("sourceOffset", offset.ToString(CultureInfo.InvariantCulture)));
 479        headers.Add(KafkaTransportHeader.Utf8("consumerGroup", _consumerGroup));
 480        headers.Add(KafkaTransportHeader.Utf8("subscriberRole", _role.ToString()));
 481        headers.Add(KafkaTransportHeader.Utf8("attempts", attempts.ToString(CultureInfo.InvariantCulture)));
 482        headers.Add(KafkaTransportHeader.Utf8("reason", reason));
 483        // Capped: these two are the only dead-letter headers whose size the failing code decides
 484        // (an exception message can quote a whole payload; a closed generic's name nests without
 485        // bound), and an uncapped one pushed the dead-letter record past message.max.bytes — a
 486        // burial that then fails on every retry, for a message that was itself within the limit.
 487        // Surrogate-aware cut (see PortableText.TruncateWellFormed).
 488        headers.Add(KafkaTransportHeader.Utf8("exceptionType", PortableText.TruncateWellFormed(exception.GetType().FullN
 489        headers.Add(KafkaTransportHeader.Utf8("exceptionMessage", PortableText.TruncateWellFormed(exception.Message, Max
 490        headers.Add(KafkaTransportHeader.Utf8("occurredAtUtc", DateTimeOffset.UtcNow.ToString("O")));
 491
 492        // The unprocessable-message discard blocks the poll thread on this (the awaiting
 493        // dispatcher's burial runs inside the detached handler task now, but keeps the same bound
 494        // so a partition is not parked on an undeliverable dead-letter topic for message.timeout.ms
 495        // per attempt either): a produce to an undeliverable dead-letter topic waits out
 496        // librdkafka's message.timeout.ms (5 min by default) PER attempt — past
 497        // max.poll.interval.ms, which evicted the consumer mid-burial and rebalanced the partition
 498        // to a peer that hit the same message: a rebalance storm at zero throughput. Bound the whole
 499        // ladder to a quarter of the poll interval; every caller already treats a failed burial as
 500        // "offset left unstored, retried after restart/rebalance".
 501        using var pollBudget = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
 502        pollBudget.CancelAfter(TimeSpan.FromTicks(_subscriberOptions.MaxPollInterval.Ticks / 4));
 503
 504        await KafkaTransportRetry.ExecuteAsync(
 505            token => _producer.PublishAsync(
 506                _topics.DeadLetterTopicFor(sourceTopic),
 507                correlationId,
 508                payload,
 509                headers,
 510                token),
 511            TransportOptions.PublishMaxAttempts,
 512            TransportOptions.PublishRetryBaseDelay,
 513            TransportOptions.PublishRetryMaxDelay,
 514            pollBudget.Token).ConfigureAwait(false);
 515    }
 516
 517    /// <summary>Runs the NotifyBackgroundFailureAsync operation.</summary>
 518    protected async ValueTask NotifyBackgroundFailureAsync(
 519        KafkaDelivery delivery,
 520        Exception exception)
 521    {
 522        var callback = _subscriberOptions.OnBackgroundFailure;
 523        if (callback is null)
 524            return;
 525
 526        try
 527        {
 528            await callback(new KafkaBackgroundFailureContext(
 529                delivery.Topic,
 530                _consumerGroup,
 531                _role.ToString(),
 532                delivery.Partition,
 533                delivery.Offset,
 534                delivery.CorrelationId,
 535                exception)).ConfigureAwait(false);
 536        }
 537        catch (Exception callbackException)
 538        {
 539            Logger.LogError(
 540                callbackException,
 541                "Kafka background failure callback failed for already-committed message {Topic}[{Partition}]@{Offset}.",
 542                delivery.Topic,
 543                delivery.Partition,
 544                delivery.Offset);
 545        }
 546    }
 547}
 548
 549/// <summary>
 550/// Ack-after-handler mode. A message's handler is started the moment it is consumed and awaited
 551/// inline for up to <see cref="KafkaSubscriberOptions.DetachHandlerAfter"/>; a handler still
 552/// running past that is <em>detached</em>: its partition is paused (Kafka's own ordering primitive
 553/// — nothing for it is fetched, nothing is buffered in-process), the handler and its retry ladder
 554/// run on, and the poll thread returns to polling. The earlier design awaited the whole handler on
 555/// the poll thread: a durable-flow step awaiting a remote response or a timer for longer than
 556/// <c>max.poll.interval.ms</c> (5 minutes by default) got the consumer evicted from its group, its
 557/// partitions rebalanced, the message redelivered to a peer that started the same work again,
 558/// and every other partition assigned to this consumer stalled behind it.
 559/// <para>
 560/// The consumer is touched only from the poll thread: detached handlers never store offsets or
 561/// resume partitions themselves. The poll loop calls <see cref="SettleCompleted"/> every tick,
 562/// which observes finished handlers exactly as the inline path would — success stores the offset,
 563/// cancellation leaves it unstored for redelivery, a burial that failed for good faults the poll
 564/// loop so nothing is ever committed past the message — then starts the next message held for the
 565/// partition, or resumes it. Disposal (the poll loop has exited by then) waits for the remaining
 566/// detached handlers and settles them before the consumer's close commits, so finished work is
 567/// not redelivered by a routine stop.
 568/// </para>
 569/// </summary>
 570internal sealed class AwaitingKafkaMessageDispatcher : KafkaMessageDispatcher
 571{
 572    private readonly TimeSpan _detachAfter;
 573    private readonly TimeSpan _faultDrainTimeout;
 574    private readonly string _topic;
 575
 576    // Poll-thread-only: the loop is the sole caller of Accept/SettleCompleted, and DisposeAsync
 577    // runs after it has exited. No lock.
 578    private readonly Dictionary<int, DetachedPartition> _detached = [];
 579
 580    /// <summary>Runs the AwaitingKafkaMessageDispatcher operation.</summary>
 581    public AwaitingKafkaMessageDispatcher(
 582        Func<KafkaDelivery, CancellationToken, Task> handler,
 583        IKafkaConsumerClient consumer,
 584        IKafkaProducerClient producer,
 585        KafkaAsyncResponseTransportOptions transportOptions,
 586        KafkaSubscriberOptions subscriberOptions,
 587        ILogger logger,
 588        string topic,
 589        string consumerGroup,
 590        KafkaSubscriberRole role)
 591        : base(handler, consumer, producer, transportOptions, subscriberOptions, logger, topic, consumerGroup, role)
 592    {
 593        _detachAfter = subscriberOptions.DetachHandlerAfter;
 594        _faultDrainTimeout = subscriberOptions.FaultDrainTimeout;
 595        _topic = topic;
 596    }
 597
 598    /// <summary>Partitions whose handler is currently detached (test observability).</summary>
 599    internal int DetachedCount => _detached.Count;
 600
 601    public override bool HasDetachedWork => _detached.Count > 0;
 602
 603    /// <summary>Handles the delivered message inline through to the offset store (the unit-test and inline-path contrac
 604    public override async Task HandleAsync(KafkaDelivery delivery, CancellationToken subscriberCancellationToken)
 605    {
 606        await SettleAsync(delivery, subscriberCancellationToken).ConfigureAwait(false);
 607        StoreOffsetAfterSettlement(delivery);
 608    }
 609
 610    /// <inheritdoc />
 611    public override void Accept(KafkaDelivery delivery, CancellationToken subscriberCancellationToken)
 612    {
 613        if (_detached.TryGetValue(delivery.Partition, out var inFlight))
 614        {
 615            // A message for a partition whose handler is still running: a rebalance handed the
 616            // partition back with its pause reset (librdkafka resets pause state on assignment),
 617            // or the client delivered a message it had fetched before the pause. Hold it behind
 618            // the running one — the partition's order is the contract — and re-assert the pause
 619            // so nothing more arrives; the hold is therefore bounded by what was already in
 620            // flight, never a queue that grows.
 621            (inFlight.Held ??= new Queue<HeldMessage>()).Enqueue(HeldMessage.For(delivery));
 622            PausePartition(delivery.Partition);
 623            return;
 624        }
 625
 626        // Started on the pool, not inline: the inline wait below is a real bound on the poll
 627        // thread's gap even for a handler whose synchronous prefix is long.
 628        var settlement = Task.Run(() => SettleAsync(delivery, subscriberCancellationToken), CancellationToken.None);
 629        if (WaitInline(settlement))
 630        {
 631            // The fast path, unchanged: settle in place and consume the next message.
 632            settlement.GetAwaiter().GetResult();
 633            StoreOffsetAfterSettlement(delivery);
 634            return;
 635        }
 636
 637        PausePartition(delivery.Partition);
 638        _detached[delivery.Partition] = new DetachedPartition(delivery, settlement, subscriberCancellationToken);
 639        Logger.LogDebug(
 640            "Kafka handler for {Topic}[{Partition}]@{Offset} is still running after {DetachAfter}; detached it and pause
 641            delivery.Topic,
 642            delivery.Partition,
 643            delivery.Offset,
 644            _detachAfter);
 645    }
 646
 647    /// <inheritdoc />
 648    public override void AcceptUnprocessable(KafkaIncomingMessage message, Exception failure, CancellationToken subscrib
 649    {
 650        if (_detached.TryGetValue(message.Partition, out var inFlight))
 651        {
 652            // Same rule as a valid delivery for the partition: the message consumed before it is
 653            // still being handled, so this one waits its turn. Settling it now would store — and
 654            // let the auto-committer commit — an offset PAST the unfinished message; a crash
 655            // after that commit skipped the unfinished message for good, and the dead-letter
 656            // copy this discard produces is of the malformed record, not of the work that was
 657            // lost. Held, it is buried and its offset stored in order, once the handler settles.
 658            (inFlight.Held ??= new Queue<HeldMessage>()).Enqueue(HeldMessage.Unprocessable(message, failure));
 659            PausePartition(message.Partition);
 660            Logger.LogDebug(
 661                "Kafka message {Topic}[{Partition}]@{Offset} could not be parsed into a delivery and is held behind the 
 662                message.Topic,
 663                message.Partition,
 664                message.Offset);
 665            return;
 666        }
 667
 668        // Nothing earlier on the partition is unresolved (every earlier message settled inline
 669        // or would be in _detached), so the discard is safe to settle at once.
 670        base.AcceptUnprocessable(message, failure, subscriberCancellationToken);
 671    }
 672
 673    /// <inheritdoc />
 674    public override void SettleCompleted()
 675    {
 676        if (_detached.Count == 0)
 677            return;
 678
 679        List<int>? finished = null;
 680        foreach (var (partition, work) in _detached)
 681        {
 682            if (work.Settlement.IsCompleted)
 683                (finished ??= []).Add(partition);
 684        }
 685
 686        if (finished is null)
 687            return;
 688
 689        foreach (var partition in finished)
 690        {
 691            var work = _detached[partition];
 692            // Removed BEFORE it is observed: a settlement that throws faults the poll loop, and the
 693            // entry must not be settled a second time by disposal.
 694            _detached.Remove(partition);
 695            work.Settlement.GetAwaiter().GetResult();
 696            StoreOffsetAfterSettlement(work.Delivery);
 697            ContinueHeld(partition, work.Held, work.SubscriberCancellationToken);
 698        }
 699    }
 700
 701    /// <summary>
 702    /// Works through the messages held behind a settled handler, in consumption order: an
 703    /// unprocessable one is dead-lettered and its offset stored right here (its turn has come —
 704    /// never ahead of the handler it was consumed behind); the first valid delivery is started
 705    /// detached with the rest still held behind it (the partition stays paused); an empty hold
 706    /// resumes the partition.
 707    /// </summary>
 708    private void ContinueHeld(int partition, Queue<HeldMessage>? held, CancellationToken subscriberCancellationToken)
 709    {
 710        while (held is { Count: > 0 })
 711        {
 712            var next = held.Dequeue();
 713            if (next.Delivery is { } delivery)
 714            {
 715                var settlement = Task.Run(() => SettleAsync(delivery, subscriberCancellationToken), CancellationToken.No
 716                _detached[partition] = new DetachedPartition(delivery, settlement, subscriberCancellationToken) { Held =
 717                return;
 718            }
 719
 720            // A burial that fails for good throws out of here and faults the poll loop, exactly
 721            // as an inline discard would; whatever is still held redelivers with the partition.
 722            DiscardUnprocessableAsync(next.Message!, next.Failure!, subscriberCancellationToken).GetAwaiter().GetResult(
 723        }
 724
 725        ResumePartition(partition);
 726    }
 727
 728    /// <summary>
 729    /// The poll loop has exited (a stop, or a fault). Detached handlers run on — the handler takes
 730    /// no cancellation token the ingress would honor — so wait for each and settle it exactly as the
 731    /// poll thread would have: an offset stored here is committed by the consumer close that
 732    /// follows, and finished work is not redelivered by a routine stop. Unbounded, as the inline
 733    /// path was (the host's shutdown budget bounds the stop as a whole). Messages still held behind
 734    /// a detached handler are dropped unstarted: their offsets are unstored, so they redeliver.
 735    /// </summary>
 736    public override async ValueTask DisposeAsync()
 737    {
 738        if (_detached.Count == 0)
 739            return;
 740
 741        Logger.LogInformation(
 742            "Waiting for {Count} detached Kafka handler(s) on {Topic} to settle before the consumer closes.",
 743            _detached.Count,
 744            _topic);
 745
 746        foreach (var (partition, work) in _detached.ToArray())
 747        {
 748            _detached.Remove(partition);
 749            await SettleAfterLoopExitAsync(work).ConfigureAwait(false);
 750        }
 751    }
 752
 753    /// <summary>
 754    /// The poll loop FAILED and the consumer is about to be closed and rebuilt. Waits at most
 755    /// <see cref="KafkaSubscriberOptions.FaultDrainTimeout"/> for the detached handlers: those
 756    /// that settled get their offsets stored, exactly as the poll thread would have (the close
 757    /// that follows commits them); the rest are abandoned — offsets unstored, so their messages
 758    /// redeliver on the rebuilt consumer while the abandoned handler may still be running — and
 759    /// observed, so each one's eventual outcome is logged instead of vanishing. Messages held
 760    /// behind a detached handler are dropped unstarted, as on a stop. The unbounded wait this
 761    /// replaces on the fault path let one long handler (a durable-flow step awaiting a remote
 762    /// response) hold the subscriber's reconnect for its whole duration, so a transient broker
 763    /// failure disabled every partition of the subscriber for as long as that step took and the
 764    /// configured reconnect policy never ran.
 765    /// </summary>
 766    public override async ValueTask TeardownAfterFaultAsync()
 767    {
 768        if (_detached.Count == 0)
 769            return;
 770
 771        Logger.LogInformation(
 772            "Kafka poll loop for {Topic} failed with {Count} detached handler(s) still running; waiting up to {FaultDrai
 773            _topic,
 774            _detached.Count,
 775            _faultDrainTimeout);
 776
 777        if (_faultDrainTimeout > TimeSpan.Zero)
 778        {
 779            var settlements = new Task[_detached.Count];
 780            var index = 0;
 781            foreach (var work in _detached.Values)
 782                settlements[index++] = work.Settlement;
 783
 784            try
 785            {
 786                await Task.WhenAll(settlements).WaitAsync(_faultDrainTimeout).ConfigureAwait(false);
 787            }
 788            catch (Exception)
 789            {
 790                // A timeout, or a settlement that faulted or was canceled: each one is observed
 791                // individually below.
 792            }
 793        }
 794
 795        foreach (var (partition, work) in _detached.ToArray())
 796        {
 797            _detached.Remove(partition);
 798            if (work.Settlement.IsCompleted)
 799            {
 800                await SettleAfterLoopExitAsync(work).ConfigureAwait(false);
 801                continue;
 802            }
 803
 804            Logger.LogWarning(
 805                "Abandoning detached Kafka handler for {Topic}[{Partition}]@{Offset}: still running {FaultDrainTimeout} 
 806                work.Delivery.Topic,
 807                work.Delivery.Partition,
 808                work.Delivery.Offset,
 809                _faultDrainTimeout);
 810            ObserveAbandoned(work);
 811        }
 812    }
 813
 814    /// <summary>
 815    /// Settles a detached handler after the poll loop has exited (a stop, or a fault whose budget
 816    /// it finished within): its offset is stored for the consumer close to commit, a cancellation
 817    /// or failure leaves it unstored so the message redelivers.
 818    /// </summary>
 819    private async Task SettleAfterLoopExitAsync(DetachedPartition work)
 820    {
 821        try
 822        {
 823            await work.Settlement.ConfigureAwait(false);
 824            StoreOffsetAfterSettlement(work.Delivery);
 825        }
 826        catch (OperationCanceledException)
 827        {
 828            Logger.LogInformation(
 829                "Detached Kafka handler for {Topic}[{Partition}]@{Offset} was canceled by the stop; its offset is left u
 830                work.Delivery.Topic,
 831                work.Delivery.Partition,
 832                work.Delivery.Offset);
 833        }
 834        catch (Exception ex)
 835        {
 836            Logger.LogError(
 837                ex,
 838                "Detached Kafka handler for {Topic}[{Partition}]@{Offset} failed while the subscriber was stopping; its 
 839                work.Delivery.Topic,
 840                work.Delivery.Partition,
 841                work.Delivery.Offset);
 842        }
 843    }
 844
 845    /// <summary>
 846    /// Logs the eventual outcome of a handler the fault teardown abandoned. It never touches the
 847    /// consumer — the one it was consumed on is closed by then — so the outcome is informational:
 848    /// the message has already been handed back to the group for redelivery.
 849    /// </summary>
 850    private void ObserveAbandoned(DetachedPartition work)
 851        => _ = work.Settlement.ContinueWith(
 852            static (settlement, state) =>
 853            {
 854                var (logger, delivery) = ((ILogger, KafkaDelivery))state!;
 855                if (settlement.IsCanceled)
 856                {
 857                    logger.LogInformation(
 858                        "Abandoned Kafka handler for {Topic}[{Partition}]@{Offset} stopped on the session's cancellation
 859                        delivery.Topic,
 860                        delivery.Partition,
 861                        delivery.Offset);
 862                }
 863                else if (settlement.IsFaulted)
 864                {
 865                    logger.LogWarning(
 866                        settlement.Exception!.GetBaseException(),
 867                        "Abandoned Kafka handler for {Topic}[{Partition}]@{Offset} failed after the consumer it was cons
 868                        delivery.Topic,
 869                        delivery.Partition,
 870                        delivery.Offset);
 871                }
 872                else
 873                {
 874                    logger.LogInformation(
 875                        "Abandoned Kafka handler for {Topic}[{Partition}]@{Offset} completed after the consumer it was c
 876                        delivery.Topic,
 877                        delivery.Partition,
 878                        delivery.Offset);
 879                }
 880            },
 881            (Logger, work.Delivery),
 882            CancellationToken.None,
 883            TaskContinuationOptions.ExecuteSynchronously,
 884            TaskScheduler.Default);
 885
 886    /// <summary>
 887    /// Runs the handler with the in-process retry ladder and, at the delivery cap, the dead-letter
 888    /// burial — everything but the offset store, which the poll thread performs once this returns.
 889    /// Returns normally when the message is settled (handled, or buried); throws on cancellation
 890    /// (offset not stored: redelivered after restart or rebalance) and when the burial fails for
 891    /// good (<see cref="KafkaDeadLetterPublishFailedException"/>).
 892    /// </summary>
 893    private async Task SettleAsync(KafkaDelivery delivery, CancellationToken subscriberCancellationToken)
 894    {
 895        var attempt = 0;
 896        while (true)
 897        {
 898            attempt++;
 899            try
 900            {
 901                await ExecuteHandlerAsync(delivery, attempt, subscriberCancellationToken).ConfigureAwait(false);
 902            }
 903            catch (OperationCanceledException) when (subscriberCancellationToken.IsCancellationRequested)
 904            {
 905                // Offset not stored: the message is redelivered after restart or rebalance.
 906                throw;
 907            }
 908            catch (Exception ex)
 909            {
 910                if (ReachedDeliveryAttempts(attempt))
 911                {
 912                    Logger.LogWarning(
 913                        ex,
 914                        "Kafka message {Topic}[{Partition}]@{Offset} reached max delivery attempts ({MaxDeliveryAttempts
 915                        delivery.Topic,
 916                        delivery.Partition,
 917                        delivery.Offset,
 918                        MaxDeliveryAttempts);
 919                    // A permanently failing dead-letter topic (UnknownTopicOrPart with auto-create
 920                    // off, an over-sized payload) burns the publish retries and then throws. That
 921                    // throw is deliberately NOT swallowed: an earlier round swallowed it, leaving
 922                    // the offset unstored and consumption running, and the next successful
 923                    // settlement on the same partition then stored a higher offset — the
 924                    // auto-committer committed past this message and a restart skipped it with no
 925                    // dead-letter copy. Faulting the subscriber (KafkaDeadLetterPublishFailedException)
 926                    // stalls the partition AT this message: the consumer closes without storing
 927                    // past it, the supervisor restarts it after its backoff, and the handler and
 928                    // burial are retried per restart — a loud, bounded-rate loop until the
 929                    // dead-letter topic is fixed, which is the at-least-once outcome.
 930                    try
 931                    {
 932                        await DeadLetterAsync(
 933                            delivery,
 934                            ex,
 935                            "handler_failed_max_attempts",
 936                            attempt,
 937                            CancellationToken.None).ConfigureAwait(false);
 938                    }
 939                    catch (Exception deadLetterException)
 940                    {
 941                        Logger.LogError(
 942                            deadLetterException,
 943                            "Failed to dead-letter Kafka message {Topic}[{Partition}]@{Offset} at the delivery cap; its 
 944                            delivery.Topic,
 945                            delivery.Partition,
 946                            delivery.Offset);
 947                        throw new KafkaDeadLetterPublishFailedException(delivery.Topic, delivery.Partition, delivery.Off
 948                    }
 949
 950                    return;
 951                }
 952
 953                // Kafka offsets cannot NACK one message, so retry in-process with backoff. This
 954                // stalls the message's partition (head-of-line), which is inherent to classic
 955                // consumer groups — and only that partition: past DetachHandlerAfter the ladder
 956                // runs detached from the poll thread.
 957                await Task.Delay(RetryBackoff(attempt), subscriberCancellationToken).ConfigureAwait(false);
 958                continue;
 959            }
 960
 961            // Settlement sits OUTSIDE the handler try (parity with the queued dispatcher and every
 962            // sibling transport): a StoreOffset failure after a successful handler — routine when a
 963            // rebalance revoked the partition mid-handler — must not be misread as a handler
 964            // failure that re-runs, or dead-letters, work that already succeeded.
 965            return;
 966        }
 967    }
 968
 969    /// <summary>
 970    /// Blocks the poll thread for at most the inline budget. <c>true</c> when the settlement task
 971    /// finished (in any state — the caller observes it); <c>false</c> when it is still running.
 972    /// </summary>
 973    private bool WaitInline(Task settlement)
 974    {
 975        if (settlement.IsCompleted)
 976            return true;
 977        if (_detachAfter <= TimeSpan.Zero)
 978            return false;
 979
 980        try
 981        {
 982            return settlement.Wait(_detachAfter);
 983        }
 984        catch (AggregateException)
 985        {
 986            // Completed, faulted: the caller re-awaits it and gets the original exception.
 987            return true;
 988        }
 989    }
 990
 991    private void PausePartition(int partition)
 992    {
 993        try
 994        {
 995            Consumer.PausePartition(_topic, partition);
 996        }
 997        catch (Exception ex)
 998        {
 999            // Not assigned any more (a rebalance took it): nothing to pause, nothing arrives for it,
 1000            // and the running handler's outcome is settled like any other when it finishes.
 1001            Logger.LogDebug(ex, "Could not pause {Topic}[{Partition}] behind its detached handler; the partition is no l
 1002        }
 1003    }
 1004
 1005    private void ResumePartition(int partition)
 1006    {
 1007        try
 1008        {
 1009            Consumer.ResumePartition(_topic, partition);
 1010        }
 1011        catch (Exception ex)
 1012        {
 1013            Logger.LogDebug(ex, "Could not resume {Topic}[{Partition}] after its detached handler settled; the partition
 1014        }
 1015    }
 1016
 1017    private sealed class DetachedPartition(KafkaDelivery delivery, Task settlement, CancellationToken subscriberCancella
 1018    {
 1019        public KafkaDelivery Delivery { get; } = delivery;
 1020        public Task Settlement { get; } = settlement;
 1021        public CancellationToken SubscriberCancellationToken { get; } = subscriberCancellationToken;
 1022
 1023        /// <summary>Messages consumed for the partition while its handler was detached, in order.</summary>
 1024        public Queue<HeldMessage>? Held { get; set; }
 1025    }
 1026
 1027    /// <summary>
 1028    /// One message consumed behind a detached handler: a valid delivery, or one that could not
 1029    /// be projected (kept with the failure that rejected it, for the dead-letter headers).
 1030    /// </summary>
 1031    private readonly record struct HeldMessage(KafkaDelivery? Delivery, KafkaIncomingMessage? Message, Exception? Failur
 1032    {
 1033        public static HeldMessage For(KafkaDelivery delivery) => new(delivery, null, null);
 1034
 1035        public static HeldMessage Unprocessable(KafkaIncomingMessage message, Exception failure) => new(null, message, f
 1036    }
 1037}
 1038
 1039internal sealed class QueuedKafkaMessageDispatcher : KafkaMessageDispatcher
 1040{
 1041    private readonly Channel<KafkaDelivery> _queue;
 1042    private readonly Task[] _workers;
 401043    private readonly CancellationTokenSource _drainCancellation = new();
 1044    private readonly TimeSpan _drainTimeout;
 1045    private readonly int _capacity;
 1046    private readonly string _topic;
 1047    private int _pendingCount;
 1048    private int _runningCount;
 1049    private int _disposeStarted;
 1050
 1051    /// <summary>Runs the QueuedKafkaMessageDispatcher operation.</summary>
 1052    public QueuedKafkaMessageDispatcher(
 1053        Func<KafkaDelivery, CancellationToken, Task> handler,
 1054        IKafkaConsumerClient consumer,
 1055        IKafkaProducerClient producer,
 1056        KafkaAsyncResponseTransportOptions transportOptions,
 1057        KafkaSubscriberOptions subscriberOptions,
 1058        ILogger logger,
 1059        string topic,
 1060        string consumerGroup,
 1061        KafkaSubscriberRole role)
 401062        : base(handler, consumer, producer, transportOptions, subscriberOptions, logger, topic, consumerGroup, role)
 1063    {
 401064        _topic = topic;
 401065        _drainTimeout = subscriberOptions.BackgroundDrainTimeout;
 401066        _capacity = subscriberOptions.BackgroundQueueCapacity;
 401067        _queue = Channel.CreateBounded<KafkaDelivery>(new BoundedChannelOptions(subscriberOptions.BackgroundQueueCapacit
 401068        {
 401069            AllowSynchronousContinuations = false,
 401070            FullMode = BoundedChannelFullMode.Wait,
 401071            SingleReader = subscriberOptions.BackgroundWorkerCount == 1,
 401072            SingleWriter = false
 401073        });
 1074
 401075        _workers = Enumerable.Range(0, subscriberOptions.BackgroundWorkerCount)
 961076            .Select(workerIndex => Task.Run(() => RunWorkerAsync(workerIndex)))
 401077            .ToArray();
 1078
 401079        Logger.LogInformation(
 401080            "Created Kafka ACK-after-enqueue dispatcher for {Topic} with {WorkerCount} worker(s), queue capacity {QueueC
 401081            _topic,
 401082            subscriberOptions.BackgroundWorkerCount,
 401083            subscriberOptions.BackgroundQueueCapacity,
 401084            _drainTimeout);
 401085    }
 1086
 481087    internal int PendingCount => Volatile.Read(ref _pendingCount);
 461088    internal int RunningCount => Volatile.Read(ref _runningCount);
 1089
 341090    public override bool CanAcceptMore => Volatile.Read(ref _pendingCount) < _capacity;
 1091
 1092    /// <summary>Handles the delivered message.</summary>
 1093    public override async Task HandleAsync(
 1094        KafkaDelivery delivery,
 1095        CancellationToken subscriberCancellationToken)
 1096    {
 881097        Interlocked.Increment(ref _pendingCount);
 881098        if (!_queue.Writer.TryWrite(delivery))
 1099        {
 1100            // The subscriber pauses partition fetching while CanAcceptMore is false, so this wait
 1101            // only covers the race between its capacity check and this write. Unlike Redis there is
 1102            // no pending-entry list to defer to: the message is already consumed, so it must be
 1103            // enqueued before the loop may continue.
 1104            try
 1105            {
 41106                await _queue.Writer.WriteAsync(delivery, subscriberCancellationToken).ConfigureAwait(false);
 01107            }
 41108            catch
 1109            {
 41110                Interlocked.Decrement(ref _pendingCount);
 41111                throw;
 1112            }
 1113        }
 1114
 1115        // The message now belongs to a background worker, which decrements _pendingCount when it
 1116        // dequeues. Do not touch the counter again here, even if the offset store below fails.
 1117        try
 1118        {
 841119            StoreOffset(delivery);
 821120        }
 21121        catch (Exception ex)
 1122        {
 21123            Logger.LogError(
 21124                ex,
 21125                "Failed to store offset for Kafka message {Topic}[{Partition}]@{Offset} after enqueue; it is being proce
 21126                delivery.Topic,
 21127                delivery.Partition,
 21128                delivery.Offset);
 21129        }
 841130    }
 1131
 1132    /// <summary>Releases resources held by this instance.</summary>
 1133    public override async ValueTask DisposeAsync()
 1134    {
 421135        if (Interlocked.Exchange(ref _disposeStarted, 1) != 0)
 21136            return;
 1137
 401138        Logger.LogInformation(
 401139            "Draining Kafka ACK-after-enqueue dispatcher for {Topic}. Pending={PendingCount}, Running={RunningCount}.",
 401140            _topic,
 401141            PendingCount,
 401142            RunningCount);
 401143        _queue.Writer.TryComplete();
 1144
 1145        try
 1146        {
 401147            await Task.WhenAll(_workers).WaitAsync(_drainTimeout).ConfigureAwait(false);
 321148            _drainCancellation.Dispose();
 321149        }
 61150        catch (TimeoutException ex)
 1151        {
 61152            _drainCancellation.Cancel();
 61153            Logger.LogWarning(
 61154                ex,
 61155                "Timed out while draining Kafka ACK-after-enqueue dispatcher for {Topic}. Pending={PendingCount}, Runnin
 61156                _topic,
 61157                PendingCount,
 61158                RunningCount);
 1159
 61160            _ = Task.WhenAll(_workers).ContinueWith(
 61161                _ => _drainCancellation.Dispose(),
 61162                CancellationToken.None,
 61163                TaskContinuationOptions.ExecuteSynchronously,
 61164                TaskScheduler.Default);
 61165        }
 21166        catch (Exception ex)
 1167        {
 1168            // A worker faulted outside its own handler guard (DB/NATS dispatcher parity). WhenAll
 1169            // only completes once every worker has finished, so the source is safe to dispose here
 1170            // — and the fault must not escape DisposeAsync and mask the real shutdown path.
 21171            Logger.LogDebug(ex, "Kafka ACK-after-enqueue dispatcher drain for {Topic} ended with an error.", _topic);
 21172            _drainCancellation.Dispose();
 21173        }
 421174    }
 1175
 1176    private async Task RunWorkerAsync(int workerIndex)
 1177    {
 2621178        await foreach (var delivery in _queue.Reader.ReadAllAsync().ConfigureAwait(false))
 1179        {
 841180            Interlocked.Decrement(ref _pendingCount);
 1181
 1182            // Once the drain budget has lapsed, STOP executing (DB/Redis/Pub-Sub parity). The
 1183            // drain token cannot stop the real handler — it is
 1184            // `_ingress.HandleWorkerMessageAsync(payload)`, whose target takes no
 1185            // CancellationToken — so past the budget the loop kept starting fresh work beyond the
 1186            // host's shutdown budget, and every entry still queued at process exit vanished with
 1187            // no record (its offset was stored at enqueue, so Kafka never redelivers it).
 841188            if (_drainCancellation.IsCancellationRequested)
 1189            {
 41190                var lapsed = new OperationCanceledException(
 41191                    "The ACK-after-enqueue drain budget lapsed before this already-committed message was handled.");
 41192                Logger.LogWarning(
 41193                    "Kafka background handler for already-committed message {Topic}[{Partition}]@{Offset} was not starte
 41194                    delivery.Topic,
 41195                    delivery.Partition,
 41196                    delivery.Offset);
 41197                await NotifyBackgroundFailureAsync(delivery, lapsed).ConfigureAwait(false);
 1198
 1199                try
 1200                {
 41201                    await DeadLetterAsync(
 41202                        delivery,
 41203                        lapsed,
 41204                        "drain_budget_lapsed_after_commit",
 41205                        0,
 41206                        CancellationToken.None).ConfigureAwait(false);
 41207                }
 01208                catch (Exception deadLetterException)
 1209                {
 01210                    Logger.LogError(
 01211                        deadLetterException,
 01212                        "Failed to dead-letter already-committed Kafka message {Topic}[{Partition}]@{Offset}.",
 01213                        delivery.Topic,
 01214                        delivery.Partition,
 01215                        delivery.Offset);
 01216                }
 1217
 1218                continue;
 1219            }
 1220
 801221            Interlocked.Increment(ref _runningCount);
 1222
 1223            try
 1224            {
 801225                await ExecuteWithRetryAsync(delivery).ConfigureAwait(false);
 741226            }
 41227            catch (OperationCanceledException ex) when (_drainCancellation.IsCancellationRequested)
 1228            {
 1229                // The drain budget lapsed with this already-committed message still unprocessed:
 1230                // Kafka will not redeliver it, so surface the drop through OnBackgroundFailure
 1231                // instead of losing it silently.
 41232                Logger.LogWarning(
 41233                    "Kafka background handler for already-committed message {Topic}[{Partition}]@{Offset} was canceled d
 41234                    delivery.Topic,
 41235                    delivery.Partition,
 41236                    delivery.Offset);
 41237                await NotifyBackgroundFailureAsync(delivery, ex).ConfigureAwait(false);
 1238            }
 1239            finally
 1240            {
 801241                Interlocked.Decrement(ref _runningCount);
 1242            }
 781243        }
 461244    }
 1245
 1246    private async Task ExecuteWithRetryAsync(KafkaDelivery delivery)
 1247    {
 801248        var attempt = 0;
 1249        while (true)
 1250        {
 921251            attempt++;
 1252            try
 1253            {
 921254                await ExecuteHandlerAsync(
 921255                    delivery,
 921256                    attempt,
 921257                    _drainCancellation.Token,
 921258                    logFailures: false).ConfigureAwait(false);
 661259                return;
 1260            }
 41261            catch (OperationCanceledException) when (_drainCancellation.IsCancellationRequested)
 1262            {
 41263                throw;
 1264            }
 1265            catch (Exception ex)
 1266            {
 1267                // Unlimited retries (0) on this path spun a background worker on an already-
 1268                // committed message forever: the bounded queue filled, the poll loop pinned the
 1269                // assignment paused, and the subscriber wedged with no dead-letter record and no
 1270                // OnBackgroundFailure, because both live inside this block. After an early ACK
 1271                // the sibling transports never retry at all, so 0 means a single attempt here.
 221272                if (MaxDeliveryAttempts <= 0 || ReachedDeliveryAttempts(attempt))
 1273                {
 101274                    Logger.LogError(
 101275                        ex,
 101276                        "Kafka background handler failed for already-committed message {Topic}[{Partition}]@{Offset} aft
 101277                        delivery.Topic,
 101278                        delivery.Partition,
 101279                        delivery.Offset,
 101280                        attempt);
 81281                    await NotifyBackgroundFailureAsync(delivery, ex).ConfigureAwait(false);
 1282
 1283                    try
 1284                    {
 81285                        await DeadLetterAsync(
 81286                            delivery,
 81287                            ex,
 81288                            "background_handler_failed_after_commit",
 81289                            attempt,
 81290                            CancellationToken.None).ConfigureAwait(false);
 61291                    }
 21292                    catch (Exception deadLetterException)
 1293                    {
 21294                        Logger.LogError(
 21295                            deadLetterException,
 21296                            "Failed to dead-letter already-committed Kafka message {Topic}[{Partition}]@{Offset}.",
 21297                            delivery.Topic,
 21298                            delivery.Partition,
 21299                            delivery.Offset);
 21300                    }
 1301
 81302                    return;
 1303                }
 1304
 121305                await Task.Delay(RetryBackoff(attempt), _drainCancellation.Token).ConfigureAwait(false);
 121306            }
 1307        }
 741308    }
 1309}
Generated by: ReportGenerator 5.5.11.0
09/22/2026 - 09:41:04
GitHub | reportgenerator.io

Methods/Properties

.ctor(System.Func`3<AsyncResponse.Transports.Kafka.KafkaDelivery,System.Threading.CancellationToken,System.Threading.Tasks.Task>,AsyncResponse.Transports.Kafka.IKafkaConsumerClient,AsyncResponse.Transports.Kafka.IKafkaProducerClient,AsyncResponse.Transports.Kafka.KafkaAsyncResponseTransportOptions,AsyncResponse.Transports.Kafka.KafkaSubscriberOptions,Microsoft.Extensions.Logging.ILogger,System.String,System.String,AsyncResponse.Transports.Kafka.KafkaSubscriberRole)
get_PendingCount()
get_RunningCount()
get_CanAcceptMore()
HandleAsync()
DisposeAsync()
RunWorkerAsync()
ExecuteWithRetryAsync()