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

Information
Class: AsyncResponse.Transports.Kafka.QueuedKafkaMessageDispatcher
Assembly: AsyncResponse.Transports.Kafka
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/Transports/AsyncResponse.Transports.Kafka/KafkaMessageDispatcher.cs
Line coverage
99%
Covered lines: 121
Uncovered lines: 1
Coverable lines: 122
Total lines: 659
Line coverage: 99.1%
Branch coverage
87%
Covered branches: 7
Total branches: 8
Branch coverage: 87.5%
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()75%4494.44%
DisposeAsync()100%22100%
RunWorkerAsync()100%11100%
ExecuteWithRetryAsync()100%22100%

File(s)

/home/runner/work/AsyncResponse/AsyncResponse/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>
 29internal abstract class KafkaMessageDispatcher : IAsyncDisposable
 30{
 31    private readonly Func<KafkaDelivery, CancellationToken, Task> _handler;
 32    private readonly KafkaSubscriberOptions _subscriberOptions;
 33    private readonly IKafkaConsumerClient _consumer;
 34    private readonly IKafkaProducerClient _producer;
 35    private readonly KafkaTransportTopicSchema _topics;
 36    private readonly string _topic;
 37    private readonly string _consumerGroup;
 38    private readonly KafkaSubscriberRole _role;
 39
 40    /// <summary>Runs the KafkaMessageDispatcher operation.</summary>
 41    protected KafkaMessageDispatcher(
 42        Func<KafkaDelivery, CancellationToken, Task> handler,
 43        IKafkaConsumerClient consumer,
 44        IKafkaProducerClient producer,
 45        KafkaAsyncResponseTransportOptions transportOptions,
 46        KafkaSubscriberOptions subscriberOptions,
 47        ILogger logger,
 48        string topic,
 49        string consumerGroup,
 50        KafkaSubscriberRole role)
 51    {
 52        _handler = handler;
 53        _consumer = consumer;
 54        _producer = producer;
 55        TransportOptions = transportOptions;
 56        _subscriberOptions = subscriberOptions;
 57        _topics = new KafkaTransportTopicSchema(transportOptions);
 58        Logger = logger;
 59        _topic = topic;
 60        _consumerGroup = consumerGroup;
 61        _role = role;
 62    }
 63
 64    protected KafkaAsyncResponseTransportOptions TransportOptions { get; }
 65    protected ILogger Logger { get; }
 66
 67    protected int MaxDeliveryAttempts => _subscriberOptions.MaxDeliveryAttempts;
 68
 69    /// <summary>Creates the configured dispatcher.</summary>
 70    public static KafkaMessageDispatcher Create(
 71        Func<KafkaDelivery, CancellationToken, Task> handler,
 72        IKafkaConsumerClient consumer,
 73        IKafkaProducerClient producer,
 74        KafkaAsyncResponseTransportOptions transportOptions,
 75        KafkaSubscriberOptions subscriberOptions,
 76        ILogger logger,
 77        string topic,
 78        string consumerGroup,
 79        KafkaSubscriberRole role)
 80    {
 81        ValidateOptions(transportOptions, subscriberOptions, role);
 82
 83        if (subscriberOptions.AckMode is KafkaAckMode.AckAfterEnqueue)
 84        {
 85            return new QueuedKafkaMessageDispatcher(
 86                handler,
 87                consumer,
 88                producer,
 89                transportOptions,
 90                subscriberOptions,
 91                logger,
 92                topic,
 93                consumerGroup,
 94                role);
 95        }
 96
 97        return new AwaitingKafkaMessageDispatcher(
 98            handler,
 99            consumer,
 100            producer,
 101            transportOptions,
 102            subscriberOptions,
 103            logger,
 104            topic,
 105            consumerGroup,
 106            role);
 107    }
 108
 109    /// <summary>Validates the supplied options.</summary>
 110    public static void ValidateOptions(
 111        KafkaAsyncResponseTransportOptions transportOptions,
 112        KafkaSubscriberOptions subscriberOptions,
 113        KafkaSubscriberRole role)
 114    {
 115        KafkaTransportOptionsValidator.ValidateCommon(transportOptions);
 116
 117        var optionPath = role is KafkaSubscriberRole.Worker
 118            ? $"{nameof(KafkaAsyncResponseTransportOptions)}.{nameof(KafkaAsyncResponseTransportOptions.WorkerSubscriber
 119            : $"{nameof(KafkaAsyncResponseTransportOptions)}.{nameof(KafkaAsyncResponseTransportOptions.ResponseSubscrib
 120
 121        if (subscriberOptions.PollTimeout <= TimeSpan.Zero)
 122            throw new InvalidOperationException($"{optionPath}.{nameof(KafkaSubscriberOptions.PollTimeout)} must be posi
 123        if (subscriberOptions.BackpressurePollDelay <= TimeSpan.Zero)
 124            throw new InvalidOperationException($"{optionPath}.{nameof(KafkaSubscriberOptions.BackpressurePollDelay)} mu
 125        if (subscriberOptions.MaxDeliveryAttempts < 0)
 126            throw new InvalidOperationException($"{optionPath}.{nameof(KafkaSubscriberOptions.MaxDeliveryAttempts)} cann
 127        if (subscriberOptions.HandlerRetryBaseDelay <= TimeSpan.Zero)
 128            throw new InvalidOperationException($"{optionPath}.{nameof(KafkaSubscriberOptions.HandlerRetryBaseDelay)} mu
 129        if (subscriberOptions.HandlerRetryMaxDelay <= TimeSpan.Zero)
 130            throw new InvalidOperationException($"{optionPath}.{nameof(KafkaSubscriberOptions.HandlerRetryMaxDelay)} mus
 131        if (subscriberOptions.HandlerRetryBaseDelay > subscriberOptions.HandlerRetryMaxDelay)
 132        {
 133            throw new InvalidOperationException(
 134                $"{optionPath}.{nameof(KafkaSubscriberOptions.HandlerRetryBaseDelay)} cannot exceed " +
 135                $"{optionPath}.{nameof(KafkaSubscriberOptions.HandlerRetryMaxDelay)}.");
 136        }
 137
 138        switch (subscriberOptions.AckMode)
 139        {
 140            case KafkaAckMode.AckAfterHandlerCompletes:
 141                return;
 142
 143            case KafkaAckMode.AckAfterEnqueue:
 144                if (subscriberOptions.BackgroundWorkerCount <= 0)
 145                {
 146                    throw new InvalidOperationException(
 147                        $"{optionPath}.{nameof(KafkaSubscriberOptions.BackgroundWorkerCount)} must be explicitly configu
 148                        $"when {nameof(KafkaSubscriberOptions.AckMode)} is {nameof(KafkaAckMode.AckAfterEnqueue)}.");
 149                }
 150
 151                if (subscriberOptions.BackgroundQueueCapacity <= 0)
 152                {
 153                    throw new InvalidOperationException(
 154                        $"{optionPath}.{nameof(KafkaSubscriberOptions.BackgroundQueueCapacity)} must be explicitly confi
 155                        $"when {nameof(KafkaSubscriberOptions.AckMode)} is {nameof(KafkaAckMode.AckAfterEnqueue)}.");
 156                }
 157
 158                if (subscriberOptions.BackgroundDrainTimeout <= TimeSpan.Zero)
 159                    throw new InvalidOperationException($"{optionPath}.{nameof(KafkaSubscriberOptions.BackgroundDrainTim
 160
 161                // Kafka subscribers spend only the background drain at shutdown; the poll loop
 162                // stops with the host token and the consumer close is not separately bounded.
 163                ShutdownBudgetValidator.Validate(
 164                    "Kafka",
 165                    $"{nameof(KafkaAsyncResponseTransportOptions)}.{nameof(KafkaAsyncResponseTransportOptions.HostShutdo
 166                    transportOptions.HostShutdownTimeout,
 167                    ($"{optionPath}.{nameof(KafkaSubscriberOptions.BackgroundDrainTimeout)}", subscriberOptions.Backgrou
 168
 169                return;
 170
 171            default:
 172                throw new InvalidOperationException(
 173                    $"{optionPath}.{nameof(KafkaSubscriberOptions.AckMode)} has unsupported value '{subscriberOptions.Ac
 174        }
 175    }
 176
 177    /// <summary>Handles the delivered message.</summary>
 178    public abstract Task HandleAsync(KafkaDelivery delivery, CancellationToken subscriberCancellationToken);
 179
 180    /// <summary>
 181    /// Whether the dispatcher can accept more deliveries right now. Awaiting dispatchers always can
 182    /// (handlers run inline); the queued dispatcher returns <c>false</c> while its bounded queue is
 183    /// saturated so the subscriber pauses partition fetching instead of buffering an unbounded
 184    /// backlog in-process.
 185    /// </summary>
 186    public virtual bool CanAcceptMore => true;
 187
 188    /// <summary>Releases resources held by this instance.</summary>
 189    public virtual ValueTask DisposeAsync() => ValueTask.CompletedTask;
 190
 191    /// <summary>Runs the ExecuteHandlerAsync operation.</summary>
 192    protected async Task ExecuteHandlerAsync(
 193        KafkaDelivery delivery,
 194        int attempt,
 195        CancellationToken cancellationToken,
 196        bool logFailures = true)
 197    {
 198        using var activity = AsyncResponseDiagnostics.StartActivity(
 199            "asyncresponse.kafka.receive",
 200            ActivityKind.Consumer,
 201            delivery.CorrelationId);
 202        activity?.SetTag("asyncresponse.transport", "kafka");
 203        activity?.SetTag("asyncresponse.kafka.role", _role.ToString());
 204        activity?.SetTag("asyncresponse.kafka.ack_mode", _subscriberOptions.AckMode.ToString());
 205        activity?.SetTag("asyncresponse.kafka.delivery_attempt", attempt);
 206        activity?.SetTag("messaging.system", "kafka");
 207        activity?.SetTag("messaging.destination.name", delivery.Topic);
 208        activity?.SetTag("messaging.kafka.consumer.group", _consumerGroup);
 209        activity?.SetTag("messaging.kafka.destination.partition", delivery.Partition);
 210        activity?.SetTag("messaging.kafka.message.offset", delivery.Offset);
 211
 212        try
 213        {
 214            await _handler(delivery, cancellationToken).ConfigureAwait(false);
 215        }
 216        catch (Exception ex)
 217        {
 218            if (logFailures)
 219            {
 220                Logger.LogError(
 221                    ex,
 222                    "Kafka message handling failed for {Topic}[{Partition}]@{Offset} (attempt {Attempt}).",
 223                    delivery.Topic,
 224                    delivery.Partition,
 225                    delivery.Offset,
 226                    attempt);
 227            }
 228
 229            AsyncResponseDiagnostics.SetError(activity, ex);
 230            throw;
 231        }
 232    }
 233
 234    /// <summary>
 235    /// Marks the delivered message resolved by storing its next offset; the consumer's
 236    /// auto-committer flushes stored offsets on the configured interval.
 237    /// </summary>
 238    protected void StoreOffset(KafkaDelivery delivery)
 239        => _consumer.StoreOffset(delivery.Topic, delivery.Partition, delivery.Offset);
 240
 241    /// <summary>Runs the ReachedDeliveryAttempts operation.</summary>
 242    protected bool ReachedDeliveryAttempts(int attempt)
 243        => MaxDeliveryAttempts > 0 && attempt >= MaxDeliveryAttempts;
 244
 245    /// <summary>Computes the delay before the next in-process handler retry.</summary>
 246    protected TimeSpan RetryBackoff(int completedAttempts)
 247        => AsyncResponseRetry.Backoff(
 248            completedAttempts,
 249            _subscriberOptions.HandlerRetryBaseDelay,
 250            _subscriberOptions.HandlerRetryMaxDelay);
 251
 252    /// <summary>
 253    /// Produces the failing message to the dead-letter topic (when enabled), preserving the
 254    /// original payload and headers and attaching failure-detail headers.
 255    /// </summary>
 256    protected async Task DeadLetterAsync(
 257        KafkaDelivery delivery,
 258        Exception exception,
 259        string reason,
 260        int attempts,
 261        CancellationToken cancellationToken)
 262        => await DeadLetterCoreAsync(
 263            delivery.Topic,
 264            delivery.Partition,
 265            delivery.Offset,
 266            Encoding.UTF8.GetBytes(delivery.Payload),
 267            delivery.Headers,
 268            delivery.CorrelationId,
 269            exception,
 270            reason,
 271            attempts,
 272            cancellationToken).ConfigureAwait(false);
 273
 274    /// <summary>
 275    /// Dead-letters (when enabled) and stores the offset of a message that could not be turned into
 276    /// a delivery — for example a foreign message with an empty payload. Without this, such a
 277    /// message would fail before <see cref="HandleAsync"/> runs on every subscriber restart and its
 278    /// partition would never advance.
 279    /// </summary>
 280    public async Task DiscardUnprocessableAsync(
 281        KafkaIncomingMessage message,
 282        Exception failure,
 283        CancellationToken cancellationToken)
 284    {
 285        Logger.LogError(
 286            failure,
 287            "Kafka message {Topic}[{Partition}]@{Offset} could not be parsed into a delivery; dead-lettering and committ
 288            message.Topic,
 289            message.Partition,
 290            message.Offset);
 291
 292        await DeadLetterCoreAsync(
 293            message.Topic,
 294            message.Partition,
 295            message.Offset,
 296            message.Payload ?? [],
 297            message.Headers,
 298            KafkaCorrelationIdExtractor.TryReadHeader(message.Headers, TransportOptions.CorrelationIdHeader),
 299            failure,
 300            "unprocessable_message",
 301            attempts: 0,
 302            cancellationToken).ConfigureAwait(false);
 303
 304        _consumer.StoreOffset(message.Topic, message.Partition, message.Offset);
 305    }
 306
 307    private async Task DeadLetterCoreAsync(
 308        string sourceTopic,
 309        int partition,
 310        long offset,
 311        byte[] payload,
 312        IReadOnlyList<KafkaTransportHeader> originalHeaders,
 313        string? correlationId,
 314        Exception exception,
 315        string reason,
 316        int attempts,
 317        CancellationToken cancellationToken)
 318    {
 319        if (!TransportOptions.DeadLetterEnabled)
 320            return;
 321
 322        var headers = new List<KafkaTransportHeader>(originalHeaders.Count + 10);
 323        headers.AddRange(originalHeaders);
 324        headers.Add(KafkaTransportHeader.Utf8("sourceTopic", sourceTopic));
 325        headers.Add(KafkaTransportHeader.Utf8("sourcePartition", partition.ToString(CultureInfo.InvariantCulture)));
 326        headers.Add(KafkaTransportHeader.Utf8("sourceOffset", offset.ToString(CultureInfo.InvariantCulture)));
 327        headers.Add(KafkaTransportHeader.Utf8("consumerGroup", _consumerGroup));
 328        headers.Add(KafkaTransportHeader.Utf8("subscriberRole", _role.ToString()));
 329        headers.Add(KafkaTransportHeader.Utf8("attempts", attempts.ToString(CultureInfo.InvariantCulture)));
 330        headers.Add(KafkaTransportHeader.Utf8("reason", reason));
 331        headers.Add(KafkaTransportHeader.Utf8("exceptionType", exception.GetType().FullName!));
 332        headers.Add(KafkaTransportHeader.Utf8("exceptionMessage", exception.Message));
 333        headers.Add(KafkaTransportHeader.Utf8("occurredAtUtc", DateTimeOffset.UtcNow.ToString("O")));
 334
 335        await KafkaTransportRetry.ExecuteAsync(
 336            token => _producer.PublishAsync(
 337                _topics.DeadLetterTopicFor(sourceTopic),
 338                correlationId,
 339                payload,
 340                headers,
 341                token),
 342            TransportOptions.PublishMaxAttempts,
 343            TransportOptions.PublishRetryBaseDelay,
 344            TransportOptions.PublishRetryMaxDelay,
 345            cancellationToken).ConfigureAwait(false);
 346    }
 347
 348    /// <summary>Runs the NotifyBackgroundFailureAsync operation.</summary>
 349    protected async ValueTask NotifyBackgroundFailureAsync(
 350        KafkaDelivery delivery,
 351        Exception exception)
 352    {
 353        var callback = _subscriberOptions.OnBackgroundFailure;
 354        if (callback is null)
 355            return;
 356
 357        try
 358        {
 359            await callback(new KafkaBackgroundFailureContext(
 360                delivery.Topic,
 361                _consumerGroup,
 362                _role.ToString(),
 363                delivery.Partition,
 364                delivery.Offset,
 365                delivery.CorrelationId,
 366                exception)).ConfigureAwait(false);
 367        }
 368        catch (Exception callbackException)
 369        {
 370            Logger.LogError(
 371                callbackException,
 372                "Kafka background failure callback failed for already-committed message {Topic}[{Partition}]@{Offset}.",
 373                delivery.Topic,
 374                delivery.Partition,
 375                delivery.Offset);
 376        }
 377    }
 378}
 379
 380internal sealed class AwaitingKafkaMessageDispatcher(
 381    Func<KafkaDelivery, CancellationToken, Task> handler,
 382    IKafkaConsumerClient consumer,
 383    IKafkaProducerClient producer,
 384    KafkaAsyncResponseTransportOptions transportOptions,
 385    KafkaSubscriberOptions subscriberOptions,
 386    ILogger logger,
 387    string topic,
 388    string consumerGroup,
 389    KafkaSubscriberRole role)
 390    : KafkaMessageDispatcher(handler, consumer, producer, transportOptions, subscriberOptions, logger, topic, consumerGr
 391{
 392    /// <summary>Handles the delivered message.</summary>
 393    public override async Task HandleAsync(
 394        KafkaDelivery delivery,
 395        CancellationToken subscriberCancellationToken)
 396    {
 397        var attempt = 0;
 398        while (true)
 399        {
 400            attempt++;
 401            try
 402            {
 403                await ExecuteHandlerAsync(delivery, attempt, subscriberCancellationToken).ConfigureAwait(false);
 404                StoreOffset(delivery);
 405                return;
 406            }
 407            catch (OperationCanceledException) when (subscriberCancellationToken.IsCancellationRequested)
 408            {
 409                // Offset not stored: the message is redelivered after restart or rebalance.
 410                throw;
 411            }
 412            catch (Exception ex)
 413            {
 414                if (ReachedDeliveryAttempts(attempt))
 415                {
 416                    Logger.LogWarning(
 417                        ex,
 418                        "Kafka message {Topic}[{Partition}]@{Offset} reached max delivery attempts ({MaxDeliveryAttempts
 419                        delivery.Topic,
 420                        delivery.Partition,
 421                        delivery.Offset,
 422                        MaxDeliveryAttempts);
 423                    await DeadLetterAsync(
 424                        delivery,
 425                        ex,
 426                        "handler_failed_max_attempts",
 427                        attempt,
 428                        CancellationToken.None).ConfigureAwait(false);
 429                    StoreOffset(delivery);
 430                    return;
 431                }
 432
 433                // Kafka offsets cannot NACK one message, so retry in-process with backoff. This
 434                // stalls the message's partition (head-of-line), which is inherent to classic
 435                // consumer groups.
 436                await Task.Delay(RetryBackoff(attempt), subscriberCancellationToken).ConfigureAwait(false);
 437            }
 438        }
 439    }
 440}
 441
 442internal sealed class QueuedKafkaMessageDispatcher : KafkaMessageDispatcher
 443{
 444    private readonly Channel<KafkaDelivery> _queue;
 445    private readonly Task[] _workers;
 3446    private readonly CancellationTokenSource _drainCancellation = new();
 447    private readonly TimeSpan _drainTimeout;
 448    private readonly int _capacity;
 449    private readonly string _topic;
 450    private int _pendingCount;
 451    private int _runningCount;
 452    private int _disposeStarted;
 453
 454    /// <summary>Runs the QueuedKafkaMessageDispatcher operation.</summary>
 455    public QueuedKafkaMessageDispatcher(
 456        Func<KafkaDelivery, CancellationToken, Task> handler,
 457        IKafkaConsumerClient consumer,
 458        IKafkaProducerClient producer,
 459        KafkaAsyncResponseTransportOptions transportOptions,
 460        KafkaSubscriberOptions subscriberOptions,
 461        ILogger logger,
 462        string topic,
 463        string consumerGroup,
 464        KafkaSubscriberRole role)
 3465        : base(handler, consumer, producer, transportOptions, subscriberOptions, logger, topic, consumerGroup, role)
 466    {
 3467        _topic = topic;
 3468        _drainTimeout = subscriberOptions.BackgroundDrainTimeout;
 3469        _capacity = subscriberOptions.BackgroundQueueCapacity;
 3470        _queue = Channel.CreateBounded<KafkaDelivery>(new BoundedChannelOptions(subscriberOptions.BackgroundQueueCapacit
 3471        {
 3472            AllowSynchronousContinuations = false,
 3473            FullMode = BoundedChannelFullMode.Wait,
 3474            SingleReader = subscriberOptions.BackgroundWorkerCount == 1,
 3475            SingleWriter = false
 3476        });
 477
 3478        _workers = Enumerable.Range(0, subscriberOptions.BackgroundWorkerCount)
 3479            .Select(workerIndex => Task.Run(() => RunWorkerAsync(workerIndex)))
 3480            .ToArray();
 481
 3482        Logger.LogInformation(
 3483            "Created Kafka ACK-after-enqueue dispatcher for {Topic} with {WorkerCount} worker(s), queue capacity {QueueC
 3484            _topic,
 3485            subscriberOptions.BackgroundWorkerCount,
 3486            subscriberOptions.BackgroundQueueCapacity,
 3487            _drainTimeout);
 3488    }
 489
 3490    internal int PendingCount => Volatile.Read(ref _pendingCount);
 3491    internal int RunningCount => Volatile.Read(ref _runningCount);
 492
 3493    public override bool CanAcceptMore => Volatile.Read(ref _pendingCount) < _capacity;
 494
 495    /// <summary>Handles the delivered message.</summary>
 496    public override async Task HandleAsync(
 497        KafkaDelivery delivery,
 498        CancellationToken subscriberCancellationToken)
 499    {
 3500        Interlocked.Increment(ref _pendingCount);
 3501        if (!_queue.Writer.TryWrite(delivery))
 502        {
 503            // The subscriber pauses partition fetching while CanAcceptMore is false, so this wait
 504            // only covers the race between its capacity check and this write. Unlike Redis there is
 505            // no pending-entry list to defer to: the message is already consumed, so it must be
 506            // enqueued before the loop may continue.
 507            try
 508            {
 2509                await _queue.Writer.WriteAsync(delivery, subscriberCancellationToken).ConfigureAwait(false);
 0510            }
 2511            catch
 512            {
 2513                Interlocked.Decrement(ref _pendingCount);
 3514                throw;
 515            }
 516        }
 517
 518        // The message now belongs to a background worker, which decrements _pendingCount when it
 519        // dequeues. Do not touch the counter again here, even if the offset store below fails.
 520        try
 521        {
 3522            StoreOffset(delivery);
 3523        }
 2524        catch (Exception ex)
 525        {
 2526            Logger.LogError(
 2527                ex,
 2528                "Failed to store offset for Kafka message {Topic}[{Partition}]@{Offset} after enqueue; it is being proce
 2529                delivery.Topic,
 2530                delivery.Partition,
 2531                delivery.Offset);
 3532        }
 3533    }
 534
 535    /// <summary>Releases resources held by this instance.</summary>
 536    public override async ValueTask DisposeAsync()
 537    {
 3538        if (Interlocked.Exchange(ref _disposeStarted, 1) != 0)
 3539            return;
 540
 3541        Logger.LogInformation(
 3542            "Draining Kafka ACK-after-enqueue dispatcher for {Topic}. Pending={PendingCount}, Running={RunningCount}.",
 3543            _topic,
 3544            PendingCount,
 3545            RunningCount);
 3546        _queue.Writer.TryComplete();
 547
 548        try
 549        {
 3550            await Task.WhenAll(_workers).WaitAsync(_drainTimeout).ConfigureAwait(false);
 3551            _drainCancellation.Dispose();
 3552        }
 2553        catch (TimeoutException ex)
 554        {
 2555            _drainCancellation.Cancel();
 2556            Logger.LogWarning(
 2557                ex,
 2558                "Timed out while draining Kafka ACK-after-enqueue dispatcher for {Topic}. Pending={PendingCount}, Runnin
 2559                _topic,
 2560                PendingCount,
 2561                RunningCount);
 562
 2563            _ = Task.WhenAll(_workers).ContinueWith(
 3564                _ => _drainCancellation.Dispose(),
 2565                CancellationToken.None,
 2566                TaskContinuationOptions.ExecuteSynchronously,
 2567                TaskScheduler.Default);
 3568        }
 3569    }
 570
 571    private async Task RunWorkerAsync(int workerIndex)
 572    {
 3573        await foreach (var delivery in _queue.Reader.ReadAllAsync().ConfigureAwait(false))
 574        {
 3575            Interlocked.Decrement(ref _pendingCount);
 3576            Interlocked.Increment(ref _runningCount);
 577
 578            try
 579            {
 3580                await ExecuteWithRetryAsync(delivery).ConfigureAwait(false);
 3581            }
 2582            catch (OperationCanceledException ex) when (_drainCancellation.IsCancellationRequested)
 583            {
 584                // The drain budget lapsed with this already-committed message still unprocessed:
 585                // Kafka will not redeliver it, so surface the drop through OnBackgroundFailure
 586                // instead of losing it silently.
 2587                Logger.LogWarning(
 2588                    "Kafka background handler for already-committed message {Topic}[{Partition}]@{Offset} was canceled d
 2589                    delivery.Topic,
 2590                    delivery.Partition,
 2591                    delivery.Offset);
 2592                await NotifyBackgroundFailureAsync(delivery, ex).ConfigureAwait(false);
 593            }
 594            finally
 595            {
 3596                Interlocked.Decrement(ref _runningCount);
 597            }
 3598        }
 3599    }
 600
 601    private async Task ExecuteWithRetryAsync(KafkaDelivery delivery)
 602    {
 3603        var attempt = 0;
 604        while (true)
 605        {
 3606            attempt++;
 607            try
 608            {
 3609                await ExecuteHandlerAsync(
 3610                    delivery,
 3611                    attempt,
 3612                    _drainCancellation.Token,
 3613                    logFailures: false).ConfigureAwait(false);
 3614                return;
 615            }
 2616            catch (OperationCanceledException) when (_drainCancellation.IsCancellationRequested)
 617            {
 2618                throw;
 619            }
 620            catch (Exception ex)
 621            {
 2622                if (ReachedDeliveryAttempts(attempt))
 623                {
 2624                    Logger.LogError(
 2625                        ex,
 2626                        "Kafka background handler failed for already-committed message {Topic}[{Partition}]@{Offset} aft
 2627                        delivery.Topic,
 2628                        delivery.Partition,
 2629                        delivery.Offset,
 2630                        attempt);
 2631                    await NotifyBackgroundFailureAsync(delivery, ex).ConfigureAwait(false);
 632
 633                    try
 634                    {
 2635                        await DeadLetterAsync(
 2636                            delivery,
 2637                            ex,
 2638                            "background_handler_failed_after_commit",
 2639                            attempt,
 2640                            CancellationToken.None).ConfigureAwait(false);
 2641                    }
 2642                    catch (Exception deadLetterException)
 643                    {
 2644                        Logger.LogError(
 2645                            deadLetterException,
 2646                            "Failed to dead-letter already-committed Kafka message {Topic}[{Partition}]@{Offset}.",
 2647                            delivery.Topic,
 2648                            delivery.Partition,
 2649                            delivery.Offset);
 2650                    }
 651
 2652                    return;
 653                }
 654
 2655                await Task.Delay(RetryBackoff(attempt), _drainCancellation.Token).ConfigureAwait(false);
 2656            }
 657        }
 3658    }
 659}
Generated by: ReportGenerator 5.5.10.0
08/03/2026 - 08:59:44
GitHub | reportgenerator.io

Methods/Properties

.ctor(System.Func<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, string, string, AsyncResponse.Transports.Kafka.KafkaSubscriberRole)
get_PendingCount()
get_RunningCount()
get_CanAcceptMore()
HandleAsync()
DisposeAsync()
RunWorkerAsync()
ExecuteWithRetryAsync()