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

Information
Class: AsyncResponse.Transports.Kafka.KafkaTransportClientDefaults
Assembly: AsyncResponse.Transports.Kafka
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/Transports/AsyncResponse.Transports.Kafka/KafkaTransportClientAdapters.cs
Line coverage
100%
Covered lines: 4
Uncovered lines: 0
Coverable lines: 4
Total lines: 324
Line coverage: 100%
Branch coverage
100%
Covered branches: 2
Total branches: 2
Branch coverage: 100%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.cctor()100%11100%
ResolveClientId(...)100%22100%

File(s)

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

#LineLine coverage
 1using Confluent.Kafka;
 2using Confluent.Kafka.Admin;
 3using System.Text;
 4
 5namespace AsyncResponse.Transports.Kafka;
 6
 7/// <summary>One Kafka message header. Values are raw bytes, per the Kafka wire model.</summary>
 8internal readonly record struct KafkaTransportHeader(string Key, byte[]? Value)
 9{
 10    public static KafkaTransportHeader Utf8(string key, string value)
 11        => new(key, Encoding.UTF8.GetBytes(value));
 12
 13    public string? ValueUtf8 => Value is null ? null : Encoding.UTF8.GetString(Value);
 14}
 15
 16/// <summary>One message pulled from a Kafka topic, decoupled from the vendor client types.</summary>
 17internal sealed record KafkaIncomingMessage(
 18    string Topic,
 19    int Partition,
 20    long Offset,
 21    byte[]? Payload,
 22    IReadOnlyList<KafkaTransportHeader> Headers);
 23
 24/// <summary>Broker coordinates assigned to a produced message.</summary>
 25internal sealed record KafkaPublishResult(string Topic, int Partition, long Offset);
 26
 27/// <summary>
 28/// Adapter seam over the Kafka producer so the publish and dead-letter paths are unit-testable
 29/// on fakes. One producer instance is shared process-wide: Kafka producers are thread-safe and
 30/// expensive to create.
 31/// </summary>
 32internal interface IKafkaProducerClient : IDisposable
 33{
 34    Task<KafkaPublishResult> PublishAsync(
 35        string topic,
 36        string? key,
 37        byte[] payload,
 38        IReadOnlyList<KafkaTransportHeader> headers,
 39        CancellationToken cancellationToken);
 40}
 41
 42/// <summary>
 43/// Adapter seam over one Kafka consumer. Not thread-safe: each hosted subscriber owns one
 44/// consumer and touches it only from its own poll loop.
 45/// </summary>
 46internal interface IKafkaConsumerClient : IDisposable
 47{
 48    /// <summary>Subscribes the consumer group member to the given topic.</summary>
 49    void Subscribe(string topic);
 50
 51    /// <summary>
 52    /// Polls for one message, waiting at most <paramref name="maxWait"/>. Returns <c>null</c> when
 53    /// no message arrived in time so the caller can re-check cancellation and backpressure state.
 54    /// </summary>
 55    KafkaIncomingMessage? Consume(TimeSpan maxWait);
 56
 57    /// <summary>
 58    /// Marks the message resolved by storing <c>offset + 1</c> for its partition. The stored
 59    /// offset is committed by the consumer's auto-committer on the configured interval, and on
 60    /// <see cref="Close"/>.
 61    /// </summary>
 62    void StoreOffset(string topic, int partition, long offset);
 63
 64    /// <summary>Pauses fetching on all currently assigned partitions (backpressure).</summary>
 65    void PauseAssignment();
 66
 67    /// <summary>Resumes fetching on all currently assigned partitions.</summary>
 68    void ResumeAssignment();
 69
 70    /// <summary>Leaves the group cleanly, committing stored offsets.</summary>
 71    void Close();
 72}
 73
 74/// <summary>Creates one consumer per hosted subscriber role.</summary>
 75internal interface IKafkaConsumerClientFactory
 76{
 77    IKafkaConsumerClient Create(KafkaSubscriberRole role);
 78}
 79
 80/// <summary>Adapter seam over the Kafka admin client for startup topic provisioning.</summary>
 81internal interface IKafkaAdminClient
 82{
 83    /// <summary>Creates any missing topics; existing topics are left untouched.</summary>
 84    Task EnsureTopicsAsync(
 85        IReadOnlyList<string> topics,
 86        int numPartitions,
 87        short replicationFactor,
 88        CancellationToken cancellationToken);
 89}
 90
 91internal sealed class KafkaProducerClientAdapter : IKafkaProducerClient
 92{
 93    private readonly KafkaAsyncResponseTransportOptions _options;
 94    private readonly Lazy<IProducer<string?, byte[]>> _producer;
 95
 96    /// <summary>Runs the KafkaProducerClientAdapter operation.</summary>
 97    public KafkaProducerClientAdapter(KafkaAsyncResponseTransportOptions options)
 98    {
 99        _options = options;
 100        // Lazy so constructing the adapter (e.g. during DI validation) never dials the broker.
 101        _producer = new Lazy<IProducer<string?, byte[]>>(CreateProducer, LazyThreadSafetyMode.ExecutionAndPublication);
 102    }
 103
 104    /// <summary>Publishes the supplied message.</summary>
 105    public async Task<KafkaPublishResult> PublishAsync(
 106        string topic,
 107        string? key,
 108        byte[] payload,
 109        IReadOnlyList<KafkaTransportHeader> headers,
 110        CancellationToken cancellationToken)
 111    {
 112        var message = new Message<string?, byte[]>
 113        {
 114            Key = key,
 115            Value = payload
 116        };
 117
 118        if (headers.Count > 0)
 119        {
 120            message.Headers = [];
 121            foreach (var header in headers)
 122                message.Headers.Add(header.Key, header.Value);
 123        }
 124
 125        var result = await _producer.Value
 126            .ProduceAsync(topic, message, cancellationToken)
 127            .ConfigureAwait(false);
 128
 129        return new KafkaPublishResult(result.Topic, result.Partition.Value, result.Offset.Value);
 130    }
 131
 132    /// <summary>Releases resources held by this instance.</summary>
 133    public void Dispose()
 134    {
 135        if (!_producer.IsValueCreated)
 136            return;
 137
 138        try
 139        {
 140            _producer.Value.Flush(_options.OperationTimeout);
 141        }
 142        catch (KafkaException)
 143        {
 144            // Best-effort drain of in-flight messages on shutdown; Dispose below always runs.
 145        }
 146
 147        _producer.Value.Dispose();
 148    }
 149
 150    private IProducer<string?, byte[]> CreateProducer()
 151    {
 152        var config = new ProducerConfig
 153        {
 154            BootstrapServers = KafkaTransportOptionsValidator.Required(
 155                _options.BootstrapServers,
 156                nameof(_options.BootstrapServers)),
 157            ClientId = KafkaTransportClientDefaults.ResolveClientId(_options),
 158            // Worker jobs must not be silently reordered or duplicated by producer-side retries.
 159            Acks = Acks.All,
 160            EnableIdempotence = true
 161        };
 162
 163        _options.ConfigureProducer?.Invoke(config);
 164        return new ProducerBuilder<string?, byte[]>(config).Build();
 165    }
 166}
 167
 168internal sealed class KafkaConsumerClientAdapter(IConsumer<string?, byte[]> _consumer) : IKafkaConsumerClient
 169{
 170    /// <summary>Subscribes the consumer group member to the given topic.</summary>
 171    public void Subscribe(string topic)
 172        => _consumer.Subscribe(topic);
 173
 174    /// <summary>Polls for one message.</summary>
 175    public KafkaIncomingMessage? Consume(TimeSpan maxWait)
 176    {
 177        var result = _consumer.Consume(maxWait);
 178        if (result is null || result.IsPartitionEOF)
 179            return null;
 180
 181        return new KafkaIncomingMessage(
 182            result.Topic,
 183            result.Partition.Value,
 184            result.Offset.Value,
 185            result.Message?.Value,
 186            ReadHeaders(result.Message?.Headers));
 187    }
 188
 189    /// <summary>Marks the message resolved by storing the next offset for its partition.</summary>
 190    public void StoreOffset(string topic, int partition, long offset)
 191        => _consumer.StoreOffset(new TopicPartitionOffset(topic, new Partition(partition), new Offset(offset + 1)));
 192
 193    /// <summary>Pauses fetching on all currently assigned partitions.</summary>
 194    public void PauseAssignment()
 195        => _consumer.Pause(_consumer.Assignment);
 196
 197    /// <summary>Resumes fetching on all currently assigned partitions.</summary>
 198    public void ResumeAssignment()
 199        => _consumer.Resume(_consumer.Assignment);
 200
 201    /// <summary>Leaves the group cleanly, committing stored offsets.</summary>
 202    public void Close()
 203        => _consumer.Close();
 204
 205    /// <summary>Releases resources held by this instance.</summary>
 206    public void Dispose()
 207        => _consumer.Dispose();
 208
 209    private static KafkaTransportHeader[] ReadHeaders(Headers? headers)
 210    {
 211        if (headers is null || headers.Count == 0)
 212            return [];
 213
 214        var result = new KafkaTransportHeader[headers.Count];
 215        for (var i = 0; i < headers.Count; i++)
 216            result[i] = new KafkaTransportHeader(headers[i].Key, headers[i].GetValueBytes());
 217
 218        return result;
 219    }
 220}
 221
 222internal sealed class KafkaConsumerClientFactory(KafkaAsyncResponseTransportOptions _options) : IKafkaConsumerClientFact
 223{
 224    /// <summary>Creates one consumer for the given subscriber role.</summary>
 225    public IKafkaConsumerClient Create(KafkaSubscriberRole role)
 226    {
 227        var config = new ConsumerConfig
 228        {
 229            BootstrapServers = KafkaTransportOptionsValidator.Required(
 230                _options.BootstrapServers,
 231                nameof(_options.BootstrapServers)),
 232            GroupId = role is KafkaSubscriberRole.Worker
 233                ? _options.WorkerConsumerGroup
 234                : _options.ResponseConsumerGroup,
 235            ClientId = $"{KafkaTransportClientDefaults.ResolveClientId(_options)}-{role.ToString().ToLowerInvariant()}",
 236            // Manual offset management: offsets are stored per resolved message (StoreOffset) and
 237            // flushed by the auto-committer, so a crash redelivers at-least-once instead of losing work.
 238            EnableAutoCommit = true,
 239            EnableAutoOffsetStore = false,
 240            AutoCommitIntervalMs = (int)Math.Max(1, _options.OffsetCommitInterval.TotalMilliseconds),
 241            // Start new consumer groups at the beginning of the topic so messages published before
 242            // the first subscriber starts are not skipped (mirrors the other transports).
 243            AutoOffsetReset = AutoOffsetReset.Earliest,
 244            EnablePartitionEof = false
 245        };
 246
 247        _options.ConfigureConsumer?.Invoke(config);
 248        return new KafkaConsumerClientAdapter(new ConsumerBuilder<string?, byte[]>(config).Build());
 249    }
 250}
 251
 252[System.Diagnostics.CodeAnalysis.ExcludeFromCodeCoverage]
 253internal sealed class KafkaAdminClientAdapter(KafkaAsyncResponseTransportOptions _options) : IKafkaAdminClient
 254{
 255    /// <summary>Creates any missing topics; existing topics are left untouched.</summary>
 256    public async Task EnsureTopicsAsync(
 257        IReadOnlyList<string> topics,
 258        int numPartitions,
 259        short replicationFactor,
 260        CancellationToken cancellationToken)
 261    {
 262        var config = new AdminClientConfig
 263        {
 264            BootstrapServers = KafkaTransportOptionsValidator.Required(
 265                _options.BootstrapServers,
 266                nameof(_options.BootstrapServers)),
 267            ClientId = $"{KafkaTransportClientDefaults.ResolveClientId(_options)}-admin"
 268        };
 269
 270        _options.ConfigureAdminClient?.Invoke(config);
 271        using var adminClient = new AdminClientBuilder(config).Build();
 272
 273        var specifications = topics
 274            .Select(topic => new TopicSpecification
 275            {
 276                Name = topic,
 277                NumPartitions = numPartitions,
 278                ReplicationFactor = replicationFactor
 279            })
 280            .ToArray();
 281
 282        try
 283        {
 284            await adminClient.CreateTopicsAsync(
 285                specifications,
 286                new CreateTopicsOptions { RequestTimeout = _options.OperationTimeout })
 287                .WaitAsync(cancellationToken)
 288                .ConfigureAwait(false);
 289        }
 290        catch (CreateTopicsException ex) when (ex.Results.All(result =>
 291            result.Error.Code is ErrorCode.NoError or ErrorCode.TopicAlreadyExists))
 292        {
 293            // The topics already exist; this is the expected path after the first app instance.
 294        }
 295    }
 296}
 297
 298internal static class KafkaTransportClientDefaults
 299{
 3300    private static readonly string GeneratedClientId = $"asyncresponse-{Environment.MachineName}-{Environment.ProcessId}
 301
 302    /// <summary>Runs the ResolveClientId operation.</summary>
 303    public static string ResolveClientId(KafkaAsyncResponseTransportOptions options)
 3304        => !string.IsNullOrWhiteSpace(options.ClientId)
 3305            ? options.ClientId!
 3306            : GeneratedClientId;
 307}
 308
 309internal static class KafkaTransportRetry
 310{
 311    /// <summary>Runs this background operation until cancellation is requested.</summary>
 312    public static Task<T> ExecuteAsync<T>(
 313        Func<CancellationToken, Task<T>> action,
 314        int maxAttempts,
 315        TimeSpan baseDelay,
 316        TimeSpan maxDelay,
 317        CancellationToken cancellationToken)
 318        => AsyncResponseRetry.ExecuteAsync(action, IsTransient, maxAttempts, baseDelay, maxDelay, cancellationToken);
 319
 320    /// <summary>Runs the IsTransient operation.</summary>
 321    public static bool IsTransient(Exception exception)
 322        => exception is TimeoutException
 323            || (exception is KafkaException kafkaException && !kafkaException.Error.IsFatal);
 324}
Generated by: ReportGenerator 5.5.10.0
08/03/2026 - 08:59:44
GitHub | reportgenerator.io

Methods/Properties

.cctor()
ResolveClientId(AsyncResponse.Transports.Kafka.KafkaAsyncResponseTransportOptions)