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

Information
Class: AsyncResponse.Transports.Kafka.KafkaSubscriberService
Assembly: AsyncResponse.Transports.Kafka
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/Transports/AsyncResponse.Transports.Kafka/KafkaSubscriberServices.cs
Line coverage
100%
Covered lines: 123
Uncovered lines: 0
Coverable lines: 123
Total lines: 291
Line coverage: 100%
Branch coverage
95%
Covered branches: 23
Total branches: 24
Branch coverage: 95.8%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%11100%
ExecuteAsync()50%22100%
RunSubscriberAsync()100%22100%
RunPollLoop(...)100%1010100%
EnsureTopicsAsync()100%22100%
CreateDelivery(...)100%88100%
CloseQuietly(...)100%11100%

File(s)

/home/runner/work/AsyncResponse/AsyncResponse/src/Transports/AsyncResponse.Transports.Kafka/KafkaSubscriberServices.cs

#LineLine coverage
 1using Microsoft.Extensions.Hosting;
 2using Microsoft.Extensions.Logging;
 3using Microsoft.Extensions.Options;
 4using System.Text;
 5
 6namespace AsyncResponse.Transports.Kafka;
 7
 8internal abstract class KafkaSubscriberService : BackgroundService
 9{
 10    private readonly IKafkaConsumerClientFactory _consumerFactory;
 11    private readonly IKafkaProducerClient _producer;
 12    private readonly IKafkaAdminClient _adminClient;
 13
 14    /// <summary>Runs the KafkaSubscriberService operation.</summary>
 315    protected KafkaSubscriberService(
 316        IOptions<KafkaAsyncResponseTransportOptions> options,
 317        IKafkaConsumerClientFactory consumerFactory,
 318        IKafkaProducerClient producer,
 319        IKafkaAdminClient adminClient,
 320        ILogger logger)
 21    {
 322        Options = options.Value;
 323        KafkaTransportOptionsValidator.ValidateCommon(Options);
 324        _consumerFactory = consumerFactory;
 325        _producer = producer;
 326        _adminClient = adminClient;
 327        Logger = logger;
 328    }
 29
 30    protected KafkaAsyncResponseTransportOptions Options { get; }
 31    protected ILogger Logger { get; }
 32
 33    protected abstract string Topic { get; }
 34    protected abstract string ConsumerGroup { get; }
 35    protected abstract KafkaSubscriberOptions SubscriberOptions { get; }
 36    protected abstract KafkaSubscriberRole SubscriberRole { get; }
 37    /// <summary>Handles the delivered message.</summary>
 38    protected abstract Task HandleMessageAsync(KafkaDelivery delivery, CancellationToken cancellationToken);
 39
 40    /// <summary>Runs this background operation until cancellation is requested.</summary>
 41    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
 42    {
 343        KafkaMessageDispatcher.ValidateOptions(Options, SubscriberOptions, SubscriberRole);
 44
 345        var failures = 0;
 346        while (!stoppingToken.IsCancellationRequested)
 47        {
 48            try
 49            {
 350                await RunSubscriberAsync(stoppingToken).ConfigureAwait(false);
 351                return;
 52            }
 253            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
 54            {
 155                return;
 56            }
 257            catch (Exception ex) when (!stoppingToken.IsCancellationRequested)
 58            {
 259                failures++;
 260                var retryDelay = AsyncResponseRetry.Backoff(
 261                    failures,
 262                    Options.SubscriberRetryBaseDelay,
 263                    Options.SubscriberRetryMaxDelay);
 264                Logger.LogWarning(
 265                    ex,
 266                    "Kafka subscriber failed for topic {Topic} ({Role}); retrying in {RetryDelay}.",
 267                    Topic,
 268                    SubscriberRole,
 269                    retryDelay);
 270                await Task.Delay(retryDelay, stoppingToken).ConfigureAwait(false);
 71            }
 72        }
 373    }
 74
 75    private async Task RunSubscriberAsync(CancellationToken stoppingToken)
 76    {
 377        if (Options.CreateTopics)
 378            await EnsureTopicsAsync(stoppingToken).ConfigureAwait(false);
 79
 380        var consumer = _consumerFactory.Create(SubscriberRole);
 81        try
 82        {
 383            consumer.Subscribe(Topic);
 384            await using var dispatcher = KafkaMessageDispatcher.Create(
 385                HandleMessageAsync,
 386                consumer,
 387                _producer,
 388                Options,
 389                SubscriberOptions,
 390                Logger,
 391                Topic,
 392                ConsumerGroup,
 393                SubscriberRole);
 94
 395            Logger.LogInformation(
 396                "Kafka subscriber started. Topic: {Topic}. Group: {ConsumerGroup}. Role: {Role}. AckMode: {AckMode}.",
 397                Topic,
 398                ConsumerGroup,
 399                SubscriberRole,
 3100                SubscriberOptions.AckMode);
 101
 102            // Consume() blocks the calling thread, so the poll loop runs on a dedicated thread
 103            // instead of starving the thread pool; the dispatcher's async work is awaited from it.
 3104            await Task.Factory.StartNew(
 3105                () => RunPollLoop(consumer, dispatcher, stoppingToken),
 3106                stoppingToken,
 3107                TaskCreationOptions.LongRunning,
 3108                TaskScheduler.Default).ConfigureAwait(false);
 109
 110            // Leaving the await-using scope drains the ACK-after-enqueue background queue before
 111            // the consumer commits its final stored offsets below.
 3112        }
 113        finally
 114        {
 3115            CloseQuietly(consumer);
 3116            consumer.Dispose();
 117        }
 3118    }
 119
 120    private void RunPollLoop(
 121        IKafkaConsumerClient consumer,
 122        KafkaMessageDispatcher dispatcher,
 123        CancellationToken stoppingToken)
 124    {
 3125        var paused = false;
 3126        while (!stoppingToken.IsCancellationRequested)
 127        {
 128            KafkaIncomingMessage? message;
 3129            if (!dispatcher.CanAcceptMore)
 130            {
 131                // Backpressure: stop fetching from the assigned partitions while the bounded
 132                // in-process queue is saturated. Keep calling Consume so the broker still sees the
 133                // consumer polling (max.poll.interval.ms) and rebalance callbacks keep firing.
 2134                if (!paused)
 135                {
 2136                    consumer.PauseAssignment();
 2137                    paused = true;
 2138                    Logger.LogDebug(
 2139                        "Kafka subscriber for {Topic} paused its assignment: the in-process queue is full.",
 2140                        Topic);
 141                }
 142
 3143                message = consumer.Consume(SubscriberOptions.BackpressurePollDelay);
 144            }
 145            else
 146            {
 3147                if (paused)
 148                {
 2149                    consumer.ResumeAssignment();
 2150                    paused = false;
 3151                    Logger.LogDebug(
 3152                        "Kafka subscriber for {Topic} resumed its assignment: in-process queue capacity freed.",
 3153                        Topic);
 154                }
 155
 3156                message = consumer.Consume(SubscriberOptions.PollTimeout);
 157            }
 158
 3159            if (message is null)
 160                continue;
 161
 162            KafkaDelivery delivery;
 163            try
 164            {
 3165                delivery = CreateDelivery(message);
 3166            }
 2167            catch (InvalidDataException ex)
 168            {
 169                // A foreign/malformed message can never be handled; dead-letter and commit it so
 170                // its partition advances instead of re-failing on every subscriber restart.
 2171                dispatcher.DiscardUnprocessableAsync(message, ex, stoppingToken).GetAwaiter().GetResult();
 3172                continue;
 173            }
 174
 3175            dispatcher.HandleAsync(delivery, stoppingToken).GetAwaiter().GetResult();
 176        }
 3177    }
 178
 179    private async Task EnsureTopicsAsync(CancellationToken cancellationToken)
 180    {
 3181        var topics = new List<string>(2) { Topic };
 3182        if (Options.DeadLetterEnabled)
 3183            topics.Add(new KafkaTransportTopicSchema(Options).DeadLetterTopicFor(Topic));
 184
 3185        await _adminClient.EnsureTopicsAsync(
 3186            topics,
 3187            Options.TopicNumPartitions,
 3188            Options.TopicReplicationFactor,
 3189            cancellationToken).ConfigureAwait(false);
 3190    }
 191
 192    private KafkaDelivery CreateDelivery(KafkaIncomingMessage message)
 193    {
 3194        var payload = message.Payload is { Length: > 0 }
 3195            ? Encoding.UTF8.GetString(message.Payload)
 3196            : null;
 3197        if (string.IsNullOrWhiteSpace(payload))
 198        {
 3199            throw new InvalidDataException(
 3200                $"Kafka message {message.Topic}[{message.Partition}]@{message.Offset} does not contain a payload.");
 201        }
 202
 3203        var correlationId = SubscriberRole is KafkaSubscriberRole.ResponseIngress
 3204            ? KafkaCorrelationIdExtractor.Extract(message.Headers, payload, Options)
 3205            : KafkaCorrelationIdExtractor.TryReadHeader(message.Headers, Options.CorrelationIdHeader);
 206
 3207        return new KafkaDelivery(
 3208            message.Topic,
 3209            message.Partition,
 3210            message.Offset,
 3211            payload,
 3212            correlationId,
 3213            message.Headers);
 214    }
 215
 216    private void CloseQuietly(IKafkaConsumerClient consumer)
 217    {
 218        try
 219        {
 220            // Commits stored offsets and leaves the group cleanly so partitions rebalance
 221            // immediately instead of waiting for the session timeout.
 3222            consumer.Close();
 3223        }
 2224        catch (Exception ex)
 225        {
 2226            Logger.LogWarning(
 2227                ex,
 2228                "Kafka consumer for topic {Topic} ({Role}) failed to close cleanly; uncommitted offsets will be redelive
 2229                Topic,
 2230                SubscriberRole);
 3231        }
 3232    }
 233}
 234
 235internal sealed class KafkaWorkerSubscriber : KafkaSubscriberService
 236{
 237    private readonly IAsyncResponseIngress _ingress;
 238    private readonly KafkaTransportTopicSchema _topics;
 239
 240    /// <summary>Runs the KafkaWorkerSubscriber operation.</summary>
 241    public KafkaWorkerSubscriber(
 242        IOptions<KafkaAsyncResponseTransportOptions> options,
 243        IKafkaConsumerClientFactory consumerFactory,
 244        IKafkaProducerClient producer,
 245        IKafkaAdminClient adminClient,
 246        IAsyncResponseIngress ingress,
 247        ILogger<KafkaWorkerSubscriber> logger)
 248        : base(options, consumerFactory, producer, adminClient, logger)
 249    {
 250        _ingress = ingress;
 251        _topics = new KafkaTransportTopicSchema(options.Value);
 252    }
 253
 254    protected override string Topic => _topics.WorkerTopic;
 255    protected override string ConsumerGroup => Options.WorkerConsumerGroup;
 256    protected override KafkaSubscriberOptions SubscriberOptions => Options.WorkerSubscriber;
 257    protected override KafkaSubscriberRole SubscriberRole => KafkaSubscriberRole.Worker;
 258
 259    /// <summary>Handles the delivered message.</summary>
 260    protected override Task HandleMessageAsync(KafkaDelivery delivery, CancellationToken cancellationToken)
 261        => _ingress.HandleWorkerMessageAsync(delivery.Payload);
 262}
 263
 264internal sealed class KafkaResponseIngressSubscriber : KafkaSubscriberService
 265{
 266    private readonly IAsyncResponseIngress _ingress;
 267    private readonly KafkaTransportTopicSchema _topics;
 268
 269    /// <summary>Runs the KafkaResponseIngressSubscriber operation.</summary>
 270    public KafkaResponseIngressSubscriber(
 271        IOptions<KafkaAsyncResponseTransportOptions> options,
 272        IKafkaConsumerClientFactory consumerFactory,
 273        IKafkaProducerClient producer,
 274        IKafkaAdminClient adminClient,
 275        IAsyncResponseIngress ingress,
 276        ILogger<KafkaResponseIngressSubscriber> logger)
 277        : base(options, consumerFactory, producer, adminClient, logger)
 278    {
 279        _ingress = ingress;
 280        _topics = new KafkaTransportTopicSchema(options.Value);
 281    }
 282
 283    protected override string Topic => _topics.ResponseTopic;
 284    protected override string ConsumerGroup => Options.ResponseConsumerGroup;
 285    protected override KafkaSubscriberOptions SubscriberOptions => Options.ResponseSubscriber;
 286    protected override KafkaSubscriberRole SubscriberRole => KafkaSubscriberRole.ResponseIngress;
 287
 288    /// <summary>Handles the delivered message.</summary>
 289    protected override Task HandleMessageAsync(KafkaDelivery delivery, CancellationToken cancellationToken)
 290        => _ingress.HandleResponseMessageAsync(delivery.Payload, delivery.CorrelationId);
 291}
Generated by: ReportGenerator 5.5.10.0
08/03/2026 - 08:59:44
GitHub | reportgenerator.io

Methods/Properties

.ctor(Microsoft.Extensions.Options.IOptions<AsyncResponse.Transports.Kafka.KafkaAsyncResponseTransportOptions>, AsyncResponse.Transports.Kafka.IKafkaConsumerClientFactory, AsyncResponse.Transports.Kafka.IKafkaProducerClient, AsyncResponse.Transports.Kafka.IKafkaAdminClient, Microsoft.Extensions.Logging.ILogger)
ExecuteAsync()
RunSubscriberAsync()
RunPollLoop(AsyncResponse.Transports.Kafka.IKafkaConsumerClient, AsyncResponse.Transports.Kafka.KafkaMessageDispatcher, System.Threading.CancellationToken)
EnsureTopicsAsync()
CreateDelivery(AsyncResponse.Transports.Kafka.KafkaIncomingMessage)
CloseQuietly(AsyncResponse.Transports.Kafka.IKafkaConsumerClient)