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

Information
Class: AsyncResponse.Transports.Kafka.KafkaIncomingMessage
Assembly: AsyncResponse.Transports.Kafka
File(s): /_/src/Transports/AsyncResponse.Transports.Kafka/KafkaTransportClientAdapters.cs
Line coverage
100%
Covered lines: 6
Uncovered lines: 0
Coverable lines: 6
Total lines: 388
Line coverage: 100%
Branch coverage
N/A
Covered branches: 0
Total branches: 0
Branch coverage: N/A
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_Topic()100%11100%
get_Partition()100%11100%
get_Offset()100%11100%
get_Payload()100%11100%
get_Headers()100%11100%

File(s)

/_/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>
 47317internal sealed record KafkaIncomingMessage(
 51318    string Topic,
 64519    int Partition,
 51320    long Offset,
 92821    byte[]? Payload,
 141122    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 — detached handlers never call it; their
 45/// completions are settled by the poll thread (and, after the loop has exited, by the
 46/// dispatcher's disposal, sequentially).
 47/// </summary>
 48internal interface IKafkaConsumerClient : IDisposable
 49{
 50    /// <summary>Subscribes the consumer group member to the given topic.</summary>
 51    void Subscribe(string topic);
 52
 53    /// <summary>
 54    /// Polls for one message, waiting at most <paramref name="maxWait"/>. Returns <c>null</c> when
 55    /// no message arrived in time so the caller can re-check cancellation and backpressure state.
 56    /// </summary>
 57    KafkaIncomingMessage? Consume(TimeSpan maxWait);
 58
 59    /// <summary>
 60    /// Marks the message resolved by storing <c>offset + 1</c> for its partition. The stored
 61    /// offset is committed by the consumer's auto-committer on the configured interval, and on
 62    /// <see cref="Close"/>.
 63    /// </summary>
 64    void StoreOffset(string topic, int partition, long offset);
 65
 66    /// <summary>Pauses fetching on all currently assigned partitions (backpressure).</summary>
 67    void PauseAssignment();
 68
 69    /// <summary>Resumes fetching on all currently assigned partitions.</summary>
 70    void ResumeAssignment();
 71
 72    /// <summary>
 73    /// Pauses fetching on one partition while a detached handler runs its message, so the
 74    /// partition's order holds with nothing buffered in-process. Throws when the partition is not
 75    /// currently assigned (revoked by a rebalance); callers treat that as informational.
 76    /// </summary>
 77    void PausePartition(string topic, int partition);
 78
 79    /// <summary>Resumes fetching on one partition once its detached handler has settled. Throws when it is no longer as
 80    void ResumePartition(string topic, int partition);
 81
 82    /// <summary>Leaves the group cleanly, committing stored offsets.</summary>
 83    void Close();
 84}
 85
 86/// <summary>Creates one consumer per hosted subscriber role.</summary>
 87internal interface IKafkaConsumerClientFactory
 88{
 89    IKafkaConsumerClient Create(KafkaSubscriberRole role);
 90}
 91
 92/// <summary>Adapter seam over the Kafka admin client for startup topic provisioning.</summary>
 93internal interface IKafkaAdminClient
 94{
 95    /// <summary>Creates any missing topics; existing topics are left untouched.</summary>
 96    Task EnsureTopicsAsync(
 97        IReadOnlyList<string> topics,
 98        int numPartitions,
 99        short replicationFactor,
 100        CancellationToken cancellationToken);
 101}
 102
 103internal sealed class KafkaProducerClientAdapter : IKafkaProducerClient
 104{
 105    private readonly KafkaAsyncResponseTransportOptions _options;
 106    private readonly object _producerGate = new();
 107    private IProducer<string?, byte[]>? _producer;
 108    private bool _disposed;
 109
 110    /// <summary>Runs the KafkaProducerClientAdapter operation.</summary>
 111    public KafkaProducerClientAdapter(KafkaAsyncResponseTransportOptions options)
 112    {
 113        _options = options;
 114    }
 115
 116    // Built lazily so constructing the adapter (e.g. during DI validation) never dials the broker,
 117    // and assigned only on SUCCESS so a faulted build attempt is not cached: this adapter is a
 118    // process-lifetime singleton, and Lazy<T>'s ExecutionAndPublication mode would rethrow one
 119    // transient construction failure on every later publish until restart. Parity with the
 120    // RabbitMQ/Pub-Sub/SQS worker transports, whose comments pin the same rule.
 121    private IProducer<string?, byte[]> Producer
 122    {
 123        get
 124        {
 125            if (Volatile.Read(ref _producer) is { } existing)
 126                return existing;
 127
 128            lock (_producerGate)
 129            {
 130                // Checked under the same gate Dispose latches under: a build racing Dispose must
 131                // either be flushed and disposed by Dispose (build won the gate) or never happen
 132                // (Dispose won) — a producer built after the latch would leak its librdkafka
 133                // threads and silently drop its buffered jobs at shutdown.
 134                ObjectDisposedException.ThrowIf(_disposed, this);
 135                return _producer ??= CreateProducer();
 136            }
 137        }
 138    }
 139
 140    /// <summary>Publishes the supplied message.</summary>
 141    public async Task<KafkaPublishResult> PublishAsync(
 142        string topic,
 143        string? key,
 144        byte[] payload,
 145        IReadOnlyList<KafkaTransportHeader> headers,
 146        CancellationToken cancellationToken)
 147    {
 148        var message = new Message<string?, byte[]>
 149        {
 150            Key = key,
 151            Value = payload
 152        };
 153
 154        if (headers.Count > 0)
 155        {
 156            message.Headers = [];
 157            foreach (var header in headers)
 158                message.Headers.Add(header.Key, header.Value);
 159        }
 160
 161        var result = await Producer
 162            .ProduceAsync(topic, message, cancellationToken)
 163            .ConfigureAwait(false);
 164
 165        return new KafkaPublishResult(result.Topic, result.Partition.Value, result.Offset.Value);
 166    }
 167
 168    /// <summary>Releases resources held by this instance.</summary>
 169    public void Dispose()
 170    {
 171        // The whole read/flush/dispose runs under the build gate: reading the field outside it
 172        // raced a concurrent build — whichever the read missed was never flushed (dropping its
 173        // buffered jobs and leaking librdkafka threads), and the reverse order left later
 174        // publishes producing onto a disposed handle.
 175        lock (_producerGate)
 176        {
 177            if (_disposed)
 178                return;
 179
 180            _disposed = true;
 181            var producer = _producer;
 182            _producer = null;
 183            if (producer is null)
 184                return;
 185
 186            try
 187            {
 188                producer.Flush(_options.OperationTimeout);
 189            }
 190            catch (KafkaException)
 191            {
 192                // Best-effort drain of in-flight messages on shutdown; Dispose below always runs.
 193            }
 194
 195            producer.Dispose();
 196        }
 197    }
 198
 199    private IProducer<string?, byte[]> CreateProducer()
 200    {
 201        var config = new ProducerConfig
 202        {
 203            BootstrapServers = KafkaTransportOptionsValidator.Required(
 204                _options.BootstrapServers,
 205                nameof(_options.BootstrapServers)),
 206            ClientId = KafkaTransportClientDefaults.ResolveClientId(_options),
 207            // Worker jobs must not be silently reordered or duplicated by producer-side retries.
 208            Acks = Acks.All,
 209            EnableIdempotence = true
 210        };
 211
 212        _options.ConfigureProducer?.Invoke(config);
 213        return new ProducerBuilder<string?, byte[]>(config).Build();
 214    }
 215}
 216
 217internal sealed class KafkaConsumerClientAdapter(IConsumer<string?, byte[]> _consumer) : IKafkaConsumerClient
 218{
 219    /// <summary>Subscribes the consumer group member to the given topic.</summary>
 220    public void Subscribe(string topic)
 221        => _consumer.Subscribe(topic);
 222
 223    /// <summary>Polls for one message.</summary>
 224    public KafkaIncomingMessage? Consume(TimeSpan maxWait)
 225    {
 226        var result = _consumer.Consume(maxWait);
 227        if (result is null || result.IsPartitionEOF)
 228            return null;
 229
 230        return new KafkaIncomingMessage(
 231            result.Topic,
 232            result.Partition.Value,
 233            result.Offset.Value,
 234            result.Message?.Value,
 235            ReadHeaders(result.Message?.Headers));
 236    }
 237
 238    /// <summary>Marks the message resolved by storing the next offset for its partition.</summary>
 239    public void StoreOffset(string topic, int partition, long offset)
 240        => _consumer.StoreOffset(new TopicPartitionOffset(topic, new Partition(partition), new Offset(offset + 1)));
 241
 242    /// <summary>Pauses fetching on all currently assigned partitions.</summary>
 243    public void PauseAssignment()
 244        => _consumer.Pause(_consumer.Assignment);
 245
 246    /// <summary>Resumes fetching on all currently assigned partitions.</summary>
 247    public void ResumeAssignment()
 248        => _consumer.Resume(_consumer.Assignment);
 249
 250    /// <summary>Pauses fetching on one partition.</summary>
 251    public void PausePartition(string topic, int partition)
 252        => _consumer.Pause([new TopicPartition(topic, new Partition(partition))]);
 253
 254    /// <summary>Resumes fetching on one partition.</summary>
 255    public void ResumePartition(string topic, int partition)
 256        => _consumer.Resume([new TopicPartition(topic, new Partition(partition))]);
 257
 258    /// <summary>Leaves the group cleanly, committing stored offsets.</summary>
 259    public void Close()
 260        => _consumer.Close();
 261
 262    /// <summary>Releases resources held by this instance.</summary>
 263    public void Dispose()
 264        => _consumer.Dispose();
 265
 266    private static KafkaTransportHeader[] ReadHeaders(Headers? headers)
 267    {
 268        if (headers is null || headers.Count == 0)
 269            return [];
 270
 271        var result = new KafkaTransportHeader[headers.Count];
 272        for (var i = 0; i < headers.Count; i++)
 273            result[i] = new KafkaTransportHeader(headers[i].Key, headers[i].GetValueBytes());
 274
 275        return result;
 276    }
 277}
 278
 279internal sealed class KafkaConsumerClientFactory(KafkaAsyncResponseTransportOptions _options) : IKafkaConsumerClientFact
 280{
 281    /// <summary>Creates one consumer for the given subscriber role.</summary>
 282    public IKafkaConsumerClient Create(KafkaSubscriberRole role)
 283    {
 284        var subscriberOptions = role is KafkaSubscriberRole.Worker
 285            ? _options.WorkerSubscriber
 286            : _options.ResponseSubscriber;
 287        var config = new ConsumerConfig
 288        {
 289            BootstrapServers = KafkaTransportOptionsValidator.Required(
 290                _options.BootstrapServers,
 291                nameof(_options.BootstrapServers)),
 292            GroupId = role is KafkaSubscriberRole.Worker
 293                ? _options.WorkerConsumerGroup
 294                : _options.ResponseConsumerGroup,
 295            ClientId = $"{KafkaTransportClientDefaults.ResolveClientId(_options)}-{role.ToString().ToLowerInvariant()}",
 296            // Manual offset management: offsets are stored per resolved message (StoreOffset) and
 297            // flushed by the auto-committer, so a crash redelivers at-least-once instead of losing work.
 298            EnableAutoCommit = true,
 299            EnableAutoOffsetStore = false,
 300            AutoCommitIntervalMs = (int)Math.Max(1, _options.OffsetCommitInterval.TotalMilliseconds),
 301            // The poll thread's longest gap (one inline handler wait of DetachHandlerAfter plus
 302            // one poll) is validated against this deadline at startup, so it is set explicitly
 303            // instead of trusting the librdkafka default to line up.
 304            MaxPollIntervalMs = (int)Math.Max(1, subscriberOptions.MaxPollInterval.TotalMilliseconds),
 305            // Start new consumer groups at the beginning of the topic so messages published before
 306            // the first subscriber starts are not skipped (mirrors the other transports).
 307            AutoOffsetReset = AutoOffsetReset.Earliest,
 308            EnablePartitionEof = false
 309        };
 310
 311        _options.ConfigureConsumer?.Invoke(config);
 312        return new KafkaConsumerClientAdapter(new ConsumerBuilder<string?, byte[]>(config).Build());
 313    }
 314}
 315
 316[System.Diagnostics.CodeAnalysis.ExcludeFromCodeCoverage]
 317internal sealed class KafkaAdminClientAdapter(KafkaAsyncResponseTransportOptions _options) : IKafkaAdminClient
 318{
 319    /// <summary>Creates any missing topics; existing topics are left untouched.</summary>
 320    public async Task EnsureTopicsAsync(
 321        IReadOnlyList<string> topics,
 322        int numPartitions,
 323        short replicationFactor,
 324        CancellationToken cancellationToken)
 325    {
 326        var config = new AdminClientConfig
 327        {
 328            BootstrapServers = KafkaTransportOptionsValidator.Required(
 329                _options.BootstrapServers,
 330                nameof(_options.BootstrapServers)),
 331            ClientId = $"{KafkaTransportClientDefaults.ResolveClientId(_options)}-admin"
 332        };
 333
 334        _options.ConfigureAdminClient?.Invoke(config);
 335        using var adminClient = new AdminClientBuilder(config).Build();
 336
 337        var specifications = topics
 338            .Select(topic => new TopicSpecification
 339            {
 340                Name = topic,
 341                NumPartitions = numPartitions,
 342                ReplicationFactor = replicationFactor
 343            })
 344            .ToArray();
 345
 346        try
 347        {
 348            await adminClient.CreateTopicsAsync(
 349                specifications,
 350                new CreateTopicsOptions { RequestTimeout = _options.OperationTimeout })
 351                .WaitAsync(cancellationToken)
 352                .ConfigureAwait(false);
 353        }
 354        catch (CreateTopicsException ex) when (ex.Results.All(result =>
 355            result.Error.Code is ErrorCode.NoError or ErrorCode.TopicAlreadyExists))
 356        {
 357            // The topics already exist; this is the expected path after the first app instance.
 358        }
 359    }
 360}
 361
 362internal static class KafkaTransportClientDefaults
 363{
 364    private static readonly string GeneratedClientId = $"asyncresponse-{Environment.MachineName}-{Environment.ProcessId}
 365
 366    /// <summary>Runs the ResolveClientId operation.</summary>
 367    public static string ResolveClientId(KafkaAsyncResponseTransportOptions options)
 368        => !string.IsNullOrWhiteSpace(options.ClientId)
 369            ? options.ClientId!
 370            : GeneratedClientId;
 371}
 372
 373internal static class KafkaTransportRetry
 374{
 375    /// <summary>Runs this background operation until cancellation is requested.</summary>
 376    public static Task<T> ExecuteAsync<T>(
 377        Func<CancellationToken, Task<T>> action,
 378        int maxAttempts,
 379        TimeSpan baseDelay,
 380        TimeSpan maxDelay,
 381        CancellationToken cancellationToken)
 382        => AsyncResponseRetry.ExecuteAsync(action, IsTransient, maxAttempts, baseDelay, maxDelay, cancellationToken);
 383
 384    /// <summary>Runs the IsTransient operation.</summary>
 385    public static bool IsTransient(Exception exception)
 386        => exception is TimeoutException
 387            || (exception is KafkaException kafkaException && !kafkaException.Error.IsFatal);
 388}
Generated by: ReportGenerator 5.5.11.0
09/22/2026 - 09:41:04
GitHub | reportgenerator.io

Methods/Properties

.ctor(System.String,System.Int32,System.Int64,System.Byte[],System.Collections.Generic.IReadOnlyList`1<AsyncResponse.Transports.Kafka.KafkaTransportHeader>)
get_Topic()
get_Partition()
get_Offset()
get_Payload()
get_Headers()