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

Information
Class: AsyncResponse.Transports.Kafka.KafkaBackgroundFailureContext
Assembly: AsyncResponse.Transports.Kafka
File(s): /_/src/Transports/AsyncResponse.Transports.Kafka/KafkaSubscriberOptions.cs
Line coverage
95%
Covered lines: 22
Uncovered lines: 1
Coverable lines: 23
Total lines: 216
Line coverage: 95.6%
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_ConsumerGroup()100%11100%
get_SubscriberRole()100%11100%
get_Partition()100%210%
get_Offset()100%11100%
get_CorrelationId()100%11100%
get_Exception()100%11100%

File(s)

/_/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 partition. A handler still running after
 14    /// <see cref="KafkaSubscriberOptions.DetachHandlerAfter"/> is detached: its partition is
 15    /// paused, the handler (retries included) runs on while the poll thread keeps polling — the
 16    /// consumer's other partitions, its <c>max.poll.interval.ms</c> liveness, and rebalance
 17    /// callbacks all continue — and the offset is stored once the handler settles. A durable flow
 18    /// awaiting a remote step or sleeping on a timer for minutes therefore no longer gets the
 19    /// consumer evicted from its group.
 20    /// </summary>
 21    AckAfterHandlerCompletes = 0,
 22
 23    /// <summary>
 24    /// Commit the offset immediately after the message is accepted into a bounded in-process
 25    /// background queue. Handler failures are retried in-process, logged, reported through
 26    /// <see cref="KafkaSubscriberOptions.OnBackgroundFailure"/>, and dead-lettered when enabled
 27    /// because the offset has already been committed. When the queue saturates, consumption is
 28    /// paused on all assigned partitions until capacity frees.
 29    /// </summary>
 30    AckAfterEnqueue = 1
 31}
 32
 33/// <summary>
 34/// Describes a handler failure that happened after a Kafka message's offset was already committed
 35/// by <see cref="KafkaAckMode.AckAfterEnqueue"/>.
 36/// </summary>
 37public sealed class KafkaBackgroundFailureContext
 38{
 1239    internal KafkaBackgroundFailureContext(
 1240        string topic,
 1241        string consumerGroup,
 1242        string subscriberRole,
 1243        int partition,
 1244        long offset,
 1245        string? correlationId,
 1246        Exception exception)
 47    {
 1248        Topic = topic;
 1249        ConsumerGroup = consumerGroup;
 1250        SubscriberRole = subscriberRole;
 1251        Partition = partition;
 1252        Offset = offset;
 1253        CorrelationId = correlationId;
 1254        Exception = exception;
 1255    }
 56
 57    /// <summary>The Kafka topic the message came from.</summary>
 258    public string Topic { get; }
 59
 60    /// <summary>The Kafka consumer group that received the message.</summary>
 261    public string ConsumerGroup { get; }
 62
 63    /// <summary>The logical subscriber role, such as <c>Worker</c> or <c>ResponseIngress</c>.</summary>
 264    public string SubscriberRole { get; }
 65
 66    /// <summary>The Kafka partition the message was read from.</summary>
 067    public int Partition { get; }
 68
 69    /// <summary>The Kafka offset of the message within its partition.</summary>
 1070    public long Offset { get; }
 71
 72    /// <summary>The AsyncResponse correlation id, when one was available.</summary>
 273    public string? CorrelationId { get; }
 74
 75    /// <summary>The exception thrown by the background handler.</summary>
 1076    public Exception Exception { get; }
 77}
 78
 79/// <summary>Per-topic Kafka subscriber behavior.</summary>
 80public sealed class KafkaSubscriberOptions
 81{
 82    /// <summary>
 83    /// Controls when a Kafka message's offset is committed. Defaults to
 84    /// <see cref="KafkaAckMode.AckAfterHandlerCompletes"/>.
 85    /// </summary>
 86    public KafkaAckMode AckMode { get; set; } = KafkaAckMode.AckAfterHandlerCompletes;
 87
 88    /// <summary>
 89    /// Maximum time one poll waits for a message before the subscriber loop re-checks cancellation
 90    /// and backpressure state. Default: <c>200ms</c>.
 91    /// </summary>
 92    public TimeSpan PollTimeout { get; set; } = TimeSpan.FromMilliseconds(200);
 93
 94    /// <summary>
 95    /// The short poll slice used while the poll thread is waiting on in-process work: capacity
 96    /// re-checks while consumption is paused because the <see cref="KafkaAckMode.AckAfterEnqueue"/>
 97    /// background queue is full, and completion checks while
 98    /// <see cref="KafkaAckMode.AckAfterHandlerCompletes"/> handlers run detached (a finished
 99    /// handler's offset is stored and its partition resumed within one slice). Default: <c>50ms</c>.
 100    /// </summary>
 101    public TimeSpan BackpressurePollDelay { get; set; } = TimeSpan.FromMilliseconds(50);
 102
 103    /// <summary>
 104    /// In <see cref="KafkaAckMode.AckAfterHandlerCompletes"/> mode, how long the poll thread waits
 105    /// for a message's handler inline before detaching it. Within the budget a fast handler settles
 106    /// exactly as before — offset stored, next message consumed, no pause. Past it the message's
 107    /// partition is paused (its order holds, nothing is buffered in-process), the handler and its
 108    /// in-process retries continue on the thread pool, and the poll thread goes back to polling:
 109    /// the consumer's other partitions keep flowing, <see cref="MaxPollInterval"/> is honored, and
 110    /// rebalance callbacks fire. The poll thread stores the offset and resumes the partition once
 111    /// the handler settles (checked every <see cref="BackpressurePollDelay"/>). Detached handlers
 112    /// for different partitions run concurrently; a stop waits for them so their offsets are
 113    /// committed. <see cref="TimeSpan.Zero"/> detaches every handler immediately. Plus
 114    /// <see cref="PollTimeout"/> this is the poll thread's longest gap, and startup validation
 115    /// requires it to fit within half of <see cref="MaxPollInterval"/>. Default: <c>1s</c>.
 116    /// </summary>
 117    public TimeSpan DetachHandlerAfter { get; set; } = TimeSpan.FromSeconds(1);
 118
 119    /// <summary>
 120    /// In <see cref="KafkaAckMode.AckAfterHandlerCompletes"/> mode, how long a subscriber whose
 121    /// poll loop <em>failed</em> (a consume error, a dropped broker connection) waits for its
 122    /// detached handlers to settle before it closes the consumer and rebuilds it. Handlers that
 123    /// settle within the budget get their offsets stored and committed by the close, exactly as
 124    /// after a stop; the rest are abandoned — their offsets stay unstored, the messages redeliver
 125    /// on the rebuilt consumer (an abandoned handler may still be running then; see the
 126    /// at-least-once notes in transport-semantics.md), and each one's eventual outcome is logged.
 127    /// Without the bound the fault teardown waited for every detached handler with no limit, so a
 128    /// transient broker failure disabled the subscriber for as long as an unrelated long handler
 129    /// — a durable-flow step awaiting a remote response — took, and the configured reconnect
 130    /// policy (<c>SubscriberRetryBaseDelay</c> → <c>SubscriberRetryMaxDelay</c>) never ran. A
 131    /// graceful stop is not bounded here; the host's shutdown budget bounds it. <see cref="TimeSpan.Zero"/>
 132    /// abandons detached handlers at once. Default: <c>5s</c>.
 133    /// </summary>
 134    public TimeSpan FaultDrainTimeout { get; set; } = TimeSpan.FromSeconds(5);
 135
 136    /// <summary>
 137    /// Maximum number of in-process delivery attempts before a failing message is produced to the
 138    /// dead-letter topic and its offset committed. Kafka offsets cannot NACK a single message, so
 139    /// retries run in-process with backoff and stall the message's partition while they run (per
 140    /// classic consumer-group semantics). <c>0</c> means unlimited retries on the
 141    /// <see cref="KafkaAckMode.AckAfterHandlerCompletes"/> path; under
 142    /// <see cref="KafkaAckMode.AckAfterEnqueue"/> the offset is already committed, so <c>0</c>
 143    /// means a single attempt before the message is dead-lettered and surfaced via
 144    /// <see cref="OnBackgroundFailure"/> (retrying a committed message forever wedged the
 145    /// background worker with no record). Attempts are counted per process delivery: a consumer
 146    /// restart before the offset commit resets the count. Default: <c>5</c>.
 147    /// </summary>
 148    public int MaxDeliveryAttempts { get; set; } = 5;
 149
 150    /// <summary>Initial delay between in-process handler retry attempts. Default: <c>100ms</c>.</summary>
 151    public TimeSpan HandlerRetryBaseDelay { get; set; } = TimeSpan.FromMilliseconds(100);
 152
 153    /// <summary>
 154    /// Maximum delay between in-process handler retry attempts. The retry ladder runs inside the
 155    /// message's handler task — detached from the poll thread past <see cref="DetachHandlerAfter"/>
 156    /// — so it stalls only that message's partition, never the consumer's group membership.
 157    /// Default: <c>5s</c>.
 158    /// </summary>
 159    public TimeSpan HandlerRetryMaxDelay { get; set; } = TimeSpan.FromSeconds(5);
 160
 161    /// <summary>
 162    /// Maximum gap between consumer polls before the broker evicts this consumer from its group
 163    /// and rebalances its partitions (the librdkafka <c>max.poll.interval.ms</c>). The poll thread's
 164    /// longest gap is one inline handler wait (<see cref="DetachHandlerAfter"/>) plus one poll
 165    /// (<see cref="PollTimeout"/>), and validation requires that sum to fit within half this
 166    /// interval; handler execution time itself is unbounded and no longer counts, because a
 167    /// handler that outlives the inline budget is detached while polling continues. Default:
 168    /// <c>5 minutes</c> (the librdkafka default).
 169    /// </summary>
 170    public TimeSpan MaxPollInterval { get; set; } = TimeSpan.FromMinutes(5);
 171
 172    /// <summary>
 173    /// Number of background workers used by <see cref="KafkaAckMode.AckAfterEnqueue"/>.
 174    /// Must be explicitly set to a positive value for early ACK mode.
 175    /// </summary>
 176    public int BackgroundWorkerCount { get; set; }
 177
 178    /// <summary>
 179    /// Maximum number of messages waiting in the background queue for
 180    /// <see cref="KafkaAckMode.AckAfterEnqueue"/>. When full, partition consumption is paused
 181    /// until capacity frees.
 182    /// </summary>
 183    public int BackgroundQueueCapacity { get; set; }
 184
 185    /// <summary>Maximum time to wait for queued/running background handlers while the hosted subscriber stops.</summary
 186    public TimeSpan BackgroundDrainTimeout { get; set; } = TimeSpan.FromSeconds(20);
 187
 188    /// <summary>
 189    /// Optional callback invoked when a background handler fails after the message's offset was
 190    /// already committed by <see cref="KafkaAckMode.AckAfterEnqueue"/>. Use it to increment
 191    /// operator-visible metrics or alert on already-committed work.
 192    /// </summary>
 193    public Func<KafkaBackgroundFailureContext, ValueTask>? OnBackgroundFailure { get; set; }
 194
 195    /// <summary>Explicitly opts this subscriber into ACK-after-enqueue behavior.</summary>
 196    public KafkaSubscriberOptions UseAckAfterEnqueue(
 197        int backgroundWorkerCount,
 198        int backgroundQueueCapacity,
 199        TimeSpan? backgroundDrainTimeout = null)
 200    {
 201        ArgumentOutOfRangeException.ThrowIfNegativeOrZero(backgroundWorkerCount);
 202        ArgumentOutOfRangeException.ThrowIfNegativeOrZero(backgroundQueueCapacity);
 203
 204        if (backgroundDrainTimeout is { } timeout && timeout <= TimeSpan.Zero)
 205            throw new ArgumentOutOfRangeException(nameof(backgroundDrainTimeout), timeout, "Drain timeout must be positi
 206
 207        AckMode = KafkaAckMode.AckAfterEnqueue;
 208        BackgroundWorkerCount = backgroundWorkerCount;
 209        BackgroundQueueCapacity = backgroundQueueCapacity;
 210
 211        if (backgroundDrainTimeout is not null)
 212            BackgroundDrainTimeout = backgroundDrainTimeout.Value;
 213
 214        return this;
 215    }
 216}
Generated by: ReportGenerator 5.5.11.0
09/22/2026 - 09:41:04
GitHub | reportgenerator.io

Methods/Properties

.ctor(System.String,System.String,System.String,System.Int32,System.Int64,System.String,System.Exception)
get_Topic()
get_ConsumerGroup()
get_SubscriberRole()
get_Partition()
get_Offset()
get_CorrelationId()
get_Exception()