< Summary - CoreWCF Coverage — PR #1766

Information
Class: CoreWCF.Channels.KafkaTransportPump
Assembly: CoreWCF.Kafka
File(s): /home/runner/work/CoreWCF/CoreWCF/src/CoreWCF.Kafka/src/CoreWCF/Channels/KafkaTransportPump.cs
Line coverage
84%
Covered lines: 189
Uncovered lines: 34
Coverable lines: 223
Total lines: 382
Line coverage: 84.7%
Branch coverage
62%
Covered branches: 61
Total branches: 97
Branch coverage: 62.8%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Cyclomatic complexity NPath complexity Sequence coverage
.ctor(...)60%101084.21%
.cctor()100%11100%
StartPumpAsync(...)100%8885.05%
>c__DisplayClass37_0/<<StartPumpAsync()66.66%1212100%
StopPumpAsync()83.33%6683.33%
GetCommitStrategyConfigValues(...)42.85%282872.72%
OnLog(...)11.11%9925%
OnError(...)0%220%
OnConsumeMessage(...)100%22100%
IncrementReceiveContextCount()100%11100%
DecrementReceiveContextCount()100%11100%
Dispose()90%1010100%
.ctor(...)100%11100%
Received(...)100%11100%
MarkAsProcessed(...)100%1010100%
Compare(...)100%11100%

File(s)

/home/runner/work/CoreWCF/CoreWCF/src/CoreWCF.Kafka/src/CoreWCF/Channels/KafkaTransportPump.cs

#LineLine coverage
 1// Licensed to the .NET Foundation under one or more agreements.
 2// The .NET Foundation licenses this file to you under the MIT license.
 3
 4using System;
 5using System.Buffers;
 6using System.Collections.Concurrent;
 7using System.Collections.Generic;
 8using System.IO.Pipelines;
 9using System.Linq;
 10using System.Net;
 11using System.Text.RegularExpressions;
 12using System.Threading;
 13using System.Threading.Tasks;
 14using Confluent.Kafka;
 15using CoreWCF.Configuration;
 16using CoreWCF.Queue.Common;
 17using CoreWCF.Runtime;
 18using Microsoft.Extensions.Logging;
 19
 20namespace CoreWCF.Channels;
 21
 22internal sealed class KafkaTransportPump : QueueTransportPump, IDisposable
 23{
 24    private readonly ILogger<KafkaTransportPump> _logger;
 25    private readonly KafkaDeliverySemantics _kafkaDeliverySemantics;
 125126    private IConsumer<byte[], byte[]> Consumer { get; set; }
 119727    private ConsumerConfig ConsumerConfig { get; set; }
 1028    internal IProducer<Null, byte[]> Producer { get; private set; }
 17529    private string Topic { get; }
 89630    internal KafkaTransportBindingElement TransportBindingElement { get; }
 31    private CountdownEvent _receiveContextCountdownEvent;
 4432    private readonly object _disposeLock = new();
 33    private readonly Uri _baseAddress;
 34    private CancellationTokenSource _cts;
 35    private AsyncManualResetEvent _mres;
 36    private bool _isStarted;
 37    private bool _isRegexSubscription;
 38    private readonly TimeSpan _closeTimeout;
 92939    internal TopicPartitionOffsetTracker OffsetTracker { get; private set; }
 40
 141    private static readonly (bool? EnableAutoCommit, bool? EnableAutoOffsetStore) s_atMostOnceConfigValues = (false, nul
 142    private static readonly (bool? EnableAutoCommit, bool? EnableAutoOffsetStore) s_atLeastOncePerMessageCommitConfigVal
 143    private static readonly (bool? EnableAutoCommit, bool? EnableAutoOffsetStore) s_atLeastOnceBatchCommitConfigValues =
 144    private static readonly Regex s_topicNameRegex =
 145        new(@"^[a-zA-Z0-9\.\-_\*\^]{1,255}$", RegexOptions.Compiled, TimeSpan.FromMilliseconds(100));
 46
 4447    public KafkaTransportPump(KafkaTransportBindingElement transportBindingElement,
 4448        ILogger<KafkaTransportPump> logger,
 4449        IServiceDispatcher serviceDispatcher, KafkaDeliverySemantics kafkaDeliverySemantics)
 50    {
 4451        _logger = logger;
 4452        _kafkaDeliverySemantics = kafkaDeliverySemantics;
 4453        Topic = WebUtility.UrlDecode(serviceDispatcher.BaseAddress.PathAndQuery.TrimStart('/'));
 4454        _isRegexSubscription = Topic.StartsWith("^");
 55
 4456        if (string.IsNullOrEmpty(Topic) || !s_topicNameRegex.IsMatch(Topic))
 57        {
 058            throw new NotSupportedException(string.Format(SR.InvalidTopicName, Topic));
 59        }
 60
 4461        if (transportBindingElement.ErrorHandlingStrategy == KafkaErrorHandlingStrategy.DeadLetterQueue)
 62        {
 263            if (string.IsNullOrEmpty(transportBindingElement.DeadLetterQueueTopic))
 64            {
 065                throw new NotSupportedException(SR.InvalidDeadLetterQueueTopicName);
 66            }
 67
 268            if (!s_topicNameRegex.IsMatch(transportBindingElement.DeadLetterQueueTopic))
 69            {
 070                throw new NotSupportedException(string.Format(SR.InvalidTopicName, Topic));
 71            }
 72        }
 4473        TransportBindingElement = (KafkaTransportBindingElement)transportBindingElement.Clone();
 4474        _baseAddress = serviceDispatcher.BaseAddress;
 4475        _closeTimeout = serviceDispatcher.Binding.CloseTimeout;
 4476    }
 77
 78    public override Task StartPumpAsync(QueueTransportContext queueTransportContext, CancellationToken token)
 79    {
 4380        _cts = CancellationTokenSource.CreateLinkedTokenSource(token);
 4381        _mres = new();
 4382        _mres.Reset();
 4383        _receiveContextCountdownEvent = new(1);
 84
 4385        _isStarted = true;
 86
 4387        ConsumerConfig = new();
 42088        foreach (var property in TransportBindingElement.Config)
 89        {
 16790            ConsumerConfig.Set(property.Key, property.Value);
 91        }
 4392        ConsumerConfig.BootstrapServers = _baseAddress.Authority;
 4393        var (enableAutoCommit, enableAutoOffsetStore) = GetCommitStrategyConfigValues(ConsumerConfig, _kafkaDeliverySema
 4394        ConsumerConfig.EnableAutoCommit = enableAutoCommit;
 4395        ConsumerConfig.EnableAutoOffsetStore = enableAutoOffsetStore;
 4396        Consumer = new ConsumerBuilder<byte[], byte[]>(ConsumerConfig)
 4397            .SetKeyDeserializer(Deserializers.ByteArray)
 4398            .SetValueDeserializer(Deserializers.ByteArray)
 4399            .SetLogHandler(OnLog)
 43100            .SetErrorHandler(OnError)
 43101            .Build();
 43102        OffsetTracker = _kafkaDeliverySemantics == KafkaDeliverySemantics.AtLeastOnce
 43103            ? new TopicPartitionOffsetTracker(Consumer, ConsumerConfig, _logger)
 43104            : null;
 105
 43106        Consumer.Subscribe(Topic);
 107
 43108        if (TransportBindingElement.ErrorHandlingStrategy == KafkaErrorHandlingStrategy.DeadLetterQueue)
 109        {
 2110            ProducerConfig producerConfig = new();
 16111            foreach (var property in TransportBindingElement.Config)
 112            {
 6113                producerConfig.Set(property.Key, property.Value);
 114            }
 2115            producerConfig.BootstrapServers = _baseAddress.Authority;
 2116            producerConfig.Acks = Acks.All;
 2117            Producer = new ProducerBuilder<Null, byte[]>(producerConfig)
 2118                .SetKeySerializer(Serializers.Null)
 2119                .SetValueSerializer(Serializers.ByteArray)
 2120                .Build();
 121        }
 122
 43123        Task.Run(async () =>
 43124        {
 755125            while (!_cts.Token.IsCancellationRequested)
 43126            {
 43127                try
 43128                {
 755129                    var consumeResult = Consumer.Consume(_cts.Token);
 712130                    if (ConsumerConfig.EnablePartitionEof == true && consumeResult.IsPartitionEOF)
 43131                    {
 0132                        continue;
 43133                    }
 43134
 712135                    _logger.LogInformation("Received message from kafka at {topicPartitionOffset}", consumeResult.TopicP
 712136                    if (_kafkaDeliverySemantics == KafkaDeliverySemantics.AtMostOnce)
 43137                    {
 269138                        Consumer.Commit(consumeResult);
 43139                    }
 443140                    else if (_kafkaDeliverySemantics == KafkaDeliverySemantics.AtLeastOnce)
 43141                    {
 443142                        OffsetTracker.Received(consumeResult);
 43143                    }
 43144
 712145                    await OnConsumeMessage(consumeResult, queueTransportContext);
 43146
 712147                }
 43148                catch (OperationCanceledException)
 43149                {
 43150                    break;
 43151                }
 0152                catch (ConsumeException e)
 43153                {
 0154                    if (e.Error.IsFatal)
 43155                    {
 0156                        _logger.LogCritical("Exit consume loop {code} {error}", e.Error.Code, e.Error.Reason);
 0157                        break;
 43158                    }
 43159
 0160                    _logger.LogError(e, "Consume error {code} {error}", e.Error.Code, e.Error.Reason);
 0161                }
 0162                catch (Exception e)
 43163                {
 0164                    _logger.LogCritical(e, "Unexpected error in consume loop; continuing");
 43165                    try
 43166                    {
 0167                        await Task.Delay(TimeSpan.FromMilliseconds(100), _cts.Token);
 0168                    }
 0169                    catch (OperationCanceledException)
 43170                    {
 0171                        break;
 43172                    }
 43173                }
 43174            }
 43175            _mres.Set();
 43176        }, _cts.Token);
 43177        return Task.CompletedTask;
 43178    }
 179
 180    public override async Task StopPumpAsync(CancellationToken token)
 181    {
 43182        if (!_isStarted)
 183        {
 0184            return;
 185        }
 186
 43187        _cts.Cancel();
 43188        await _mres.WaitAsync(token);
 43189        _cts.Dispose();
 190
 43191        if (ConsumerConfig.EnableAutoCommit == true)
 192        {
 193            // When EnableAutoCommit is true, offset are either manually stored locally or automatically (if EnableAutoO
 194            // Then a background librdkafka thread will commit them at AutoCommitIntervalMs frequency which defaults to 
 195            // Thus we should give AutoCommitIntervalMs time before closing the consumer
 7196            await Task.Delay(TimeSpan.FromMilliseconds(ConsumerConfig.AutoCommitIntervalMs ?? 5000));
 197        }
 198
 43199        _receiveContextCountdownEvent.Signal();
 43200        using CancellationTokenSource closeCts = new (_closeTimeout);
 201        try
 202        {
 43203            _receiveContextCountdownEvent.Wait(closeCts.Token);
 43204        }
 0205        catch (OperationCanceledException e)
 206        {
 207            // no-op
 208            // Consumer.Close and Producer.Flush will allow to gracefully handle that service stops
 0209        }
 210
 43211        _receiveContextCountdownEvent.Dispose();
 43212        if (TransportBindingElement.ErrorHandlingStrategy == KafkaErrorHandlingStrategy.DeadLetterQueue)
 213        {
 2214            Producer.Flush(closeCts.Token);
 215        }
 216
 43217        Consumer.Close();
 43218    }
 219
 220    private static (bool? EnableAutoCommit, bool? EnableAutoOffsetStore) GetCommitStrategyConfigValues(ConsumerConfig co
 43221        (kafkaDeliverySemantics, consumerConfig) switch
 43222        {
 43223            // KafkaBinding
 33224            (KafkaDeliverySemantics.AtMostOnce, { EnableAutoCommit: null, EnableAutoOffsetStore: null }) => s_atMostOnce
 7225            (KafkaDeliverySemantics.AtLeastOnce, { EnableAutoCommit: null, EnableAutoOffsetStore: null } ) => s_atLeastO
 43226            // CustomBinding
 0227            (KafkaDeliverySemantics.AtMostOnce, { EnableAutoCommit: false, EnableAutoOffsetStore: null }) => s_atMostOnc
 0228            (KafkaDeliverySemantics.AtLeastOnce, { EnableAutoCommit: true, EnableAutoOffsetStore: false } ) => s_atLeast
 3229            (KafkaDeliverySemantics.AtLeastOnce, { EnableAutoCommit: false, EnableAutoOffsetStore: null }) => s_atLeastO
 0230            _ => throw new NotSupportedException(string.Format(SR.InvalidKafkaConfiguration, kafkaDeliverySemantics, con
 43231        };
 232
 233    private void OnLog(IConsumer<byte[], byte[]> consumer, LogMessage logMessage)
 234    {
 235        const string format = "{0}:{1}";
 2921236        switch (logMessage.Level)
 237        {
 238            case SyslogLevel.Debug:
 2921239                _logger.LogDebug(format, logMessage.Name, logMessage.Message);
 2921240                break;
 241            case SyslogLevel.Notice:
 242            case SyslogLevel.Info:
 0243                _logger.LogInformation(format, logMessage.Name, logMessage.Message);
 0244                break;
 245            case SyslogLevel.Warning:
 0246                _logger.LogWarning(format, logMessage.Name, logMessage.Message);
 0247                break;
 248            case SyslogLevel.Error:
 0249                _logger.LogError(format, logMessage.Name, logMessage.Message);
 0250                break;
 251            case SyslogLevel.Alert:
 252            case SyslogLevel.Critical:
 253            case SyslogLevel.Emergency:
 0254                _logger.LogCritical(format, logMessage.Name, logMessage.Message);
 0255                break;
 256            default:
 0257                throw new ArgumentOutOfRangeException(nameof(logMessage.Level));
 258        }
 259    }
 260
 261    private void OnError(IConsumer<byte[], byte[]> consumer, Error error)
 262    {
 0263        if (error.IsFatal)
 264        {
 0265            _cts.Cancel();
 266        }
 0267    }
 268
 269    private Task OnConsumeMessage(ConsumeResult<byte[], byte[]> consumeResult,
 270        QueueTransportContext queueTransportContext)
 271    {
 713272        var receiveContext = new KafkaReceiveContext(consumeResult, this);
 713273        var context = new KafkaMessageContext
 713274        {
 713275            IsRegexSubscription = _isRegexSubscription,
 713276            ReceiveContext = receiveContext,
 713277            QueueTransportContext = queueTransportContext,
 713278            LocalAddress = new EndpointAddress(queueTransportContext.ServiceDispatcher.BaseAddress),
 713279            QueueMessageReader = PipeReader.Create(new ReadOnlySequence<byte>(consumeResult.Message.Value ?? Array.Empty
 713280            Properties =
 713281            {
 713282                [KafkaMessageProperty.Name] = new KafkaMessageProperty(consumeResult)
 713283            }
 713284        };
 285
 713286        return queueTransportContext.QueueMessageDispatcher(context);
 287    }
 288
 289    internal void IncrementReceiveContextCount()
 290    {
 713291        _receiveContextCountdownEvent.AddCount();
 713292    }
 293
 294    internal void DecrementReceiveContextCount()
 295    {
 713296        _receiveContextCountdownEvent.Signal();
 713297    }
 298
 299    public void Dispose()
 300    {
 44301        lock (_disposeLock)
 302        {
 44303            if (TransportBindingElement.ErrorHandlingStrategy == KafkaErrorHandlingStrategy.DeadLetterQueue)
 304            {
 2305                Producer?.Dispose();
 2306                Producer = null;
 307            }
 44308            Consumer?.Dispose();
 44309            Consumer = null;
 44310            _cts?.Dispose();
 44311            _mres?.Dispose();
 43312        }
 44313    }
 314
 315    internal class TopicPartitionOffsetTracker
 316    {
 10317        private readonly ConcurrentDictionary<TopicPartition, SortedDictionary<ConsumeResult<byte[], byte[]>, bool>> _to
 318        private readonly IConsumer<byte[], byte[]> _consumer;
 319        private readonly ConsumerConfig _config;
 320        private readonly ILogger<KafkaTransportPump> _logger;
 321
 10322        public TopicPartitionOffsetTracker(IConsumer<byte[], byte[]> consumer, ConsumerConfig config, ILogger<KafkaTrans
 323        {
 10324            _consumer = consumer;
 10325            _config = config;
 10326            _logger = logger;
 10327        }
 328
 329        public void Received(ConsumeResult<byte[], byte[]> consumeResult)
 330        {
 443331            SortedDictionary<ConsumeResult<byte[], byte[]>, bool> sortedDictionary =
 443332                _topicPartitions.GetOrAdd(consumeResult.TopicPartition, new SortedDictionary<ConsumeResult<byte[], byte[
 443333            lock (sortedDictionary)
 334            {
 443335                sortedDictionary.Add(consumeResult, false);
 443336            }
 443337        }
 338
 339        public void MarkAsProcessed(ConsumeResult<byte[], byte[]> consumeResult)
 340        {
 443341            ConsumeResult<byte[], byte[]> highestConsumeResult = null;
 443342            SortedDictionary<ConsumeResult<byte[], byte[]>, bool> sortedDictionary = _topicPartitions[consumeResult.Topi
 443343            lock (sortedDictionary)
 344            {
 443345                sortedDictionary[consumeResult] = true;
 346                KeyValuePair<ConsumeResult<byte[], byte[]>, bool> first;
 886347                while (sortedDictionary.Count > 0 && (first = sortedDictionary.First()).Value)
 348                {
 443349                    highestConsumeResult = first.Key;
 443350                    sortedDictionary.Remove(first.Key);
 351                }
 352
 443353                if (highestConsumeResult != null)
 354                {
 397355                    if (_config.EnableAutoCommit == false)
 356                    {
 102357                        _consumer.Commit(highestConsumeResult);
 102358                        _logger.LogDebug("Commit {topicPartitionOffset}",
 102359                            highestConsumeResult.TopicPartitionOffset);
 360                    }
 295361                    else if (_config.EnableAutoOffsetStore == false)
 362                    {
 295363                        _consumer.StoreOffset(highestConsumeResult);
 295364                        _logger.LogDebug("StoreOffsets {topicPartitionOffset}",
 295365                            highestConsumeResult.TopicPartitionOffset);
 366                    }
 367                }
 341368            }
 443369        }
 370
 371        private class ConsumeResultComparer : IComparer<ConsumeResult<byte[], byte[]>>
 372        {
 444373            public static ConsumeResultComparer Default { get; } = new();
 374
 375            public int Compare(ConsumeResult<byte[], byte[]> x, ConsumeResult<byte[], byte[]> y)
 376            {
 1586377                Fx.AssertAndThrow(x.TopicPartition == y.TopicPartition, "ConsumeResult instances must be from the same T
 1586378                return x.Offset.Value.CompareTo(y.Offset.Value);
 379            }
 380        }
 381    }
 382}
Generated by: ReportGenerator 5.5.11.0
08/09/2026 - 10:15:48
GitHub | reportgenerator.io

Methods/Properties

Consumer()
ConsumerConfig()
Producer()
Topic()
TransportBindingElement()
.ctor(CoreWCF.Channels.KafkaTransportBindingElement,Microsoft.Extensions.Logging.ILogger`1<CoreWCF.Channels.KafkaTransportPump>,CoreWCF.Configuration.IServiceDispatcher,CoreWCF.Channels.KafkaDeliverySemantics)
OffsetTracker()
.cctor()
StartPumpAsync(CoreWCF.Queue.Common.QueueTransportContext,System.Threading.CancellationToken)
>c__DisplayClass37_0/<<StartPumpAsync()
StopPumpAsync()
GetCommitStrategyConfigValues(Confluent.Kafka.ConsumerConfig,CoreWCF.Channels.KafkaDeliverySemantics)
OnLog(Confluent.Kafka.IConsumer`2<System.Byte[],System.Byte[]>,Confluent.Kafka.LogMessage)
OnError(Confluent.Kafka.IConsumer`2<System.Byte[],System.Byte[]>,Confluent.Kafka.Error)
OnConsumeMessage(Confluent.Kafka.ConsumeResult`2<System.Byte[],System.Byte[]>,CoreWCF.Queue.Common.QueueTransportContext)
IncrementReceiveContextCount()
DecrementReceiveContextCount()
Dispose()
.ctor(Confluent.Kafka.IConsumer`2<System.Byte[],System.Byte[]>,Confluent.Kafka.ConsumerConfig,Microsoft.Extensions.Logging.ILogger`1<CoreWCF.Channels.KafkaTransportPump>)
Received(Confluent.Kafka.ConsumeResult`2<System.Byte[],System.Byte[]>)
MarkAsProcessed(Confluent.Kafka.ConsumeResult`2<System.Byte[],System.Byte[]>)
Default()
Compare(Confluent.Kafka.ConsumeResult`2<System.Byte[],System.Byte[]>,Confluent.Kafka.ConsumeResult`2<System.Byte[],System.Byte[]>)