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

Information
Class: AsyncResponse.Transports.Kafka.KafkaBackgroundFailureContext
Assembly: AsyncResponse.Transports.Kafka
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/Transports/AsyncResponse.Transports.Kafka/KafkaSubscriberOptions.cs
Line coverage
100%
Covered lines: 16
Uncovered lines: 0
Coverable lines: 16
Total lines: 158
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%

File(s)

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

#LineLine coverage
 1namespace AsyncResponse.Transports.Kafka;
 2
 3/// <summary>
 4/// Controls when a Kafka message's offset is committed relative to AsyncResponse handling.
 5/// </summary>
 6public enum KafkaAckMode
 7{
 8    /// <summary>
 9    /// Commit the offset only after the AsyncResponse handler completes. If the handler throws, the
 10    /// message is retried in-process with backoff (Kafka offsets cannot NACK a single message);
 11    /// after <see cref="KafkaSubscriberOptions.MaxDeliveryAttempts"/> the message is produced to
 12    /// the dead-letter topic and its offset is committed so the partition keeps moving. Messages
 13    /// are processed serially per assignment, preserving per-partition ordering.
 14    /// </summary>
 15    AckAfterHandlerCompletes = 0,
 16
 17    /// <summary>
 18    /// Commit the offset immediately after the message is accepted into a bounded in-process
 19    /// background queue. Handler failures are retried in-process, logged, reported through
 20    /// <see cref="KafkaSubscriberOptions.OnBackgroundFailure"/>, and dead-lettered when enabled
 21    /// because the offset has already been committed. When the queue saturates, consumption is
 22    /// paused on all assigned partitions until capacity frees.
 23    /// </summary>
 24    AckAfterEnqueue = 1
 25}
 26
 27/// <summary>
 28/// Describes a handler failure that happened after a Kafka message's offset was already committed
 29/// by <see cref="KafkaAckMode.AckAfterEnqueue"/>.
 30/// </summary>
 31public sealed class KafkaBackgroundFailureContext
 32{
 233    internal KafkaBackgroundFailureContext(
 234        string topic,
 235        string consumerGroup,
 236        string subscriberRole,
 237        int partition,
 238        long offset,
 239        string? correlationId,
 240        Exception exception)
 41    {
 342        Topic = topic;
 343        ConsumerGroup = consumerGroup;
 344        SubscriberRole = subscriberRole;
 345        Partition = partition;
 346        Offset = offset;
 347        CorrelationId = correlationId;
 348        Exception = exception;
 349    }
 50
 51    /// <summary>The Kafka topic the message came from.</summary>
 52    public string Topic { get; }
 53
 54    /// <summary>The Kafka consumer group that received the message.</summary>
 55    public string ConsumerGroup { get; }
 56
 57    /// <summary>The logical subscriber role, such as <c>Worker</c> or <c>ResponseIngress</c>.</summary>
 58    public string SubscriberRole { get; }
 59
 60    /// <summary>The Kafka partition the message was read from.</summary>
 61    public int Partition { get; }
 62
 63    /// <summary>The Kafka offset of the message within its partition.</summary>
 64    public long Offset { get; }
 65
 66    /// <summary>The AsyncResponse correlation id, when one was available.</summary>
 67    public string? CorrelationId { get; }
 68
 69    /// <summary>The exception thrown by the background handler.</summary>
 70    public Exception Exception { get; }
 71}
 72
 73/// <summary>Per-topic Kafka subscriber behavior.</summary>
 74public sealed class KafkaSubscriberOptions
 75{
 76    /// <summary>
 77    /// Controls when a Kafka message's offset is committed. Defaults to
 78    /// <see cref="KafkaAckMode.AckAfterHandlerCompletes"/>.
 79    /// </summary>
 80    public KafkaAckMode AckMode { get; set; } = KafkaAckMode.AckAfterHandlerCompletes;
 81
 82    /// <summary>
 83    /// Maximum time one poll waits for a message before the subscriber loop re-checks cancellation
 84    /// and backpressure state. Default: <c>200ms</c>.
 85    /// </summary>
 86    public TimeSpan PollTimeout { get; set; } = TimeSpan.FromMilliseconds(200);
 87
 88    /// <summary>
 89    /// Delay between capacity re-checks while consumption is paused because the
 90    /// <see cref="KafkaAckMode.AckAfterEnqueue"/> background queue is full. Default: <c>50ms</c>.
 91    /// </summary>
 92    public TimeSpan BackpressurePollDelay { get; set; } = TimeSpan.FromMilliseconds(50);
 93
 94    /// <summary>
 95    /// Maximum number of in-process delivery attempts before a failing message is produced to the
 96    /// dead-letter topic and its offset committed. Kafka offsets cannot NACK a single message, so
 97    /// retries run in-process with backoff and stall the message's partition while they run (per
 98    /// classic consumer-group semantics). <c>0</c> means unlimited retries. Attempts are counted
 99    /// per process delivery: a consumer restart before the offset commit resets the count.
 100    /// Default: <c>5</c>.
 101    /// </summary>
 102    public int MaxDeliveryAttempts { get; set; } = 5;
 103
 104    /// <summary>Initial delay between in-process handler retry attempts. Default: <c>100ms</c>.</summary>
 105    public TimeSpan HandlerRetryBaseDelay { get; set; } = TimeSpan.FromMilliseconds(100);
 106
 107    /// <summary>
 108    /// Maximum delay between in-process handler retry attempts. Keep the total retry budget well
 109    /// below the consumer's <c>max.poll.interval.ms</c> (default 5 minutes) or the broker will
 110    /// evict the consumer from its group mid-retry. Default: <c>5s</c>.
 111    /// </summary>
 112    public TimeSpan HandlerRetryMaxDelay { get; set; } = TimeSpan.FromSeconds(5);
 113
 114    /// <summary>
 115    /// Number of background workers used by <see cref="KafkaAckMode.AckAfterEnqueue"/>.
 116    /// Must be explicitly set to a positive value for early ACK mode.
 117    /// </summary>
 118    public int BackgroundWorkerCount { get; set; }
 119
 120    /// <summary>
 121    /// Maximum number of messages waiting in the background queue for
 122    /// <see cref="KafkaAckMode.AckAfterEnqueue"/>. When full, partition consumption is paused
 123    /// until capacity frees.
 124    /// </summary>
 125    public int BackgroundQueueCapacity { get; set; }
 126
 127    /// <summary>Maximum time to wait for queued/running background handlers while the hosted subscriber stops.</summary
 128    public TimeSpan BackgroundDrainTimeout { get; set; } = TimeSpan.FromSeconds(20);
 129
 130    /// <summary>
 131    /// Optional callback invoked when a background handler fails after the message's offset was
 132    /// already committed by <see cref="KafkaAckMode.AckAfterEnqueue"/>. Use it to increment
 133    /// operator-visible metrics or alert on already-committed work.
 134    /// </summary>
 135    public Func<KafkaBackgroundFailureContext, ValueTask>? OnBackgroundFailure { get; set; }
 136
 137    /// <summary>Explicitly opts this subscriber into ACK-after-enqueue behavior.</summary>
 138    public KafkaSubscriberOptions UseAckAfterEnqueue(
 139        int backgroundWorkerCount,
 140        int backgroundQueueCapacity,
 141        TimeSpan? backgroundDrainTimeout = null)
 142    {
 143        ArgumentOutOfRangeException.ThrowIfNegativeOrZero(backgroundWorkerCount);
 144        ArgumentOutOfRangeException.ThrowIfNegativeOrZero(backgroundQueueCapacity);
 145
 146        if (backgroundDrainTimeout is { } timeout && timeout <= TimeSpan.Zero)
 147            throw new ArgumentOutOfRangeException(nameof(backgroundDrainTimeout), timeout, "Drain timeout must be positi
 148
 149        AckMode = KafkaAckMode.AckAfterEnqueue;
 150        BackgroundWorkerCount = backgroundWorkerCount;
 151        BackgroundQueueCapacity = backgroundQueueCapacity;
 152
 153        if (backgroundDrainTimeout is not null)
 154            BackgroundDrainTimeout = backgroundDrainTimeout.Value;
 155
 156        return this;
 157    }
 158}
Generated by: ReportGenerator 5.5.10.0
08/03/2026 - 08:59:44
GitHub | reportgenerator.io

Methods/Properties

.ctor(string, string, string, int, long, string, System.Exception)