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

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

File(s)

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

#LineLine coverage
 1using Confluent.Kafka;
 2
 3namespace AsyncResponse.Transports.Kafka;
 4
 5/// <summary>
 6/// Options for the Apache Kafka AsyncResponse transport. Built on classic consumer groups: manual
 7/// offset management, in-process bounded retries (offsets cannot NACK a single message), and
 8/// dead-letter topics. Ordering is per-partition, so consumer parallelism equals the partition
 9/// count and a slow message delays its partition (head-of-line blocking); size
 10/// <see cref="TopicNumPartitions"/> accordingly.
 11/// </summary>
 12public sealed class KafkaAsyncResponseTransportOptions
 13{
 14    public const string TransportName = "kafka";
 15
 16    /// <summary>
 17    /// Comma-separated Kafka bootstrap servers (for example <c>broker-1:9092,broker-2:9092</c>).
 18    /// Required.
 19    /// </summary>
 20    public string? BootstrapServers { get; set; }
 21
 22    /// <summary>
 23    /// Optional Kafka <c>client.id</c> applied to the producer, consumers, and admin client. When
 24    /// null, the package generates a stable process-local id containing machine name and process id.
 25    /// </summary>
 26    public string? ClientId { get; set; }
 27
 28    /// <summary>
 29    /// Prefix used when a topic name is not explicitly configured. The default worker topic is
 30    /// <c>{TopicPrefix}.transport.worker</c> and the default response topic is
 31    /// <c>{TopicPrefix}.transport.response</c>. Use a unique prefix per app/environment when several
 32    /// deployments share one cluster.
 33    /// </summary>
 334    public string TopicPrefix { get; set; } = "asyncresponse";
 35
 36    /// <summary>
 37    /// Kafka topic used by <see cref="KafkaWorkerTransport"/> to publish worker jobs. When null,
 38    /// <see cref="TopicPrefix"/> determines the topic name.
 39    /// </summary>
 40    public string? WorkerTopic { get; set; }
 41
 42    /// <summary>Consumer group used by the hosted worker subscriber.</summary>
 343    public string WorkerConsumerGroup { get; set; } = "asyncresponse-workers";
 44
 45    /// <summary>Worker topic handling options.</summary>
 346    public KafkaSubscriberOptions WorkerSubscriber { get; } = new();
 47
 48    /// <summary>
 49    /// Kafka topic remote systems can produce response payloads to. The hosted response-ingress
 50    /// subscriber reads this topic and forwards payloads into <see cref="IAsyncResponseIngress"/>.
 51    /// When null, <see cref="TopicPrefix"/> determines the topic name.
 52    /// </summary>
 53    public string? ResponseTopic { get; set; }
 54
 55    /// <summary>Consumer group used by the hosted response-ingress subscriber.</summary>
 356    public string ResponseConsumerGroup { get; set; } = "asyncresponse-responses";
 57
 58    /// <summary>Response topic handling options.</summary>
 359    public KafkaSubscriberOptions ResponseSubscriber { get; } = new();
 60
 61    /// <summary>
 62    /// Creates the worker, response, and dead-letter topics on subscriber startup using
 63    /// <see cref="TopicNumPartitions"/> and <see cref="TopicReplicationFactor"/>. Existing topics
 64    /// are left untouched. Disable when your infra team owns topic provisioning.
 65    /// </summary>
 366    public bool CreateTopics { get; set; } = true;
 67
 68    /// <summary>
 69    /// Partition count used when <see cref="CreateTopics"/> provisions a missing topic. Partitions
 70    /// are the unit of consumer parallelism and per-message ordering. <c>-1</c> uses the broker
 71    /// default (<c>num.partitions</c>). Default: <c>8</c>.
 72    /// </summary>
 373    public int TopicNumPartitions { get; set; } = 8;
 74
 75    /// <summary>
 76    /// Replication factor used when <see cref="CreateTopics"/> provisions a missing topic.
 77    /// <c>-1</c> uses the broker default (<c>default.replication.factor</c>), which is correct for
 78    /// single-broker development clusters and lets production clusters keep their own policy.
 79    /// </summary>
 380    public short TopicReplicationFactor { get; set; } = -1;
 81
 82    /// <summary>
 83    /// Enables dead-lettering when a message exhausts
 84    /// <see cref="KafkaSubscriberOptions.MaxDeliveryAttempts"/> in-process retries, a background
 85    /// handler fails after early ACK, or a message cannot be parsed into a delivery. The failing
 86    /// message is produced to the dead-letter topic with failure-detail headers and its offset is
 87    /// committed so the partition keeps moving.
 88    /// </summary>
 389    public bool DeadLetterEnabled { get; set; } = true;
 90
 91    /// <summary>
 92    /// Explicit dead-letter topic receiving poison messages from all subscribed topics. When null,
 93    /// each source topic dead-letters to <c>{sourceTopic}{DeadLetterTopicSuffix}</c>.
 94    /// </summary>
 95    public string? DeadLetterTopic { get; set; }
 96
 97    /// <summary>
 98    /// Suffix appended to a source topic to derive its dead-letter topic when
 99    /// <see cref="DeadLetterTopic"/> is not configured. Default: <c>.deadletter</c>.
 100    /// </summary>
 3101    public string DeadLetterTopicSuffix { get; set; } = ".deadletter";
 102
 103    /// <summary>The logical reply target name used by <c>WithReplyTarget()</c>. Default: <c>default</c>.</summary>
 3104    public string DefaultReplyTargetName { get; set; } = "default";
 105
 106    /// <summary>
 107    /// Named reply targets exposed to Core through <see cref="IAsyncResponseReplyTargetProvider"/>.
 108    /// When empty, the resolved <see cref="ResponseTopic"/> becomes the default target.
 109    /// </summary>
 3110    public Dictionary<string, KafkaReplyTargetOptions> ReplyTargets { get; } = new(StringComparer.Ordinal);
 111
 112    /// <summary>
 113    /// Kafka message header carrying the AsyncResponse correlation id. Response messages may omit
 114    /// it when the id is present in the JSON body via <see cref="CorrelationIdJsonPaths"/>.
 115    /// </summary>
 3116    public string CorrelationIdHeader { get; set; } = "correlationId";
 117
 118    /// <summary>
 119    /// JSON paths inspected when a response message does not carry the correlation id in
 120    /// <see cref="CorrelationIdHeader"/>. Paths are case-insensitive and support nested JSON strings.
 121    /// </summary>
 122    public string[] CorrelationIdJsonPaths { get; set; } =
 3123    [
 3124        "CorrelationId",
 3125        "CustomParameters",
 3126        "CustomParameters.CorrelationId",
 3127        "PubSubParams.CustomParameters",
 3128        "PubSubParams.CustomParameters.CorrelationId",
 3129        "DagJsonParameters.CorrelationId"
 3130    ];
 131
 132    /// <summary>
 133    /// How often stored offsets are auto-committed to the broker. Subscribers store an offset only
 134    /// once its message is fully resolved (handled, enqueued for early ACK, or dead-lettered), so a
 135    /// crash inside this window redelivers at-least-once rather than losing work. Maps to the
 136    /// consumer's <c>auto.commit.interval.ms</c>. Default: <c>5s</c>.
 137    /// </summary>
 3138    public TimeSpan OffsetCommitInterval { get; set; } = TimeSpan.FromSeconds(5);
 139
 140    /// <summary>Timeout applied to administrative operations such as topic creation.</summary>
 3141    public TimeSpan OperationTimeout { get; set; } = TimeSpan.FromSeconds(10);
 142
 143    /// <summary>Maximum attempts for Kafka produce calls. Set to 1 to disable publish retries.</summary>
 3144    public int PublishMaxAttempts { get; set; } = 3;
 145
 146    /// <summary>Initial delay before retrying a failed produce call.</summary>
 3147    public TimeSpan PublishRetryBaseDelay { get; set; } = TimeSpan.FromMilliseconds(50);
 148
 149    /// <summary>Maximum delay between produce retry attempts.</summary>
 3150    public TimeSpan PublishRetryMaxDelay { get; set; } = TimeSpan.FromSeconds(1);
 151
 152    /// <summary>Initial delay after a subscriber loop Kafka failure.</summary>
 3153    public TimeSpan SubscriberRetryBaseDelay { get; set; } = TimeSpan.FromMilliseconds(100);
 154
 155    /// <summary>Maximum delay after repeated subscriber loop Kafka failures.</summary>
 3156    public TimeSpan SubscriberRetryMaxDelay { get; set; } = TimeSpan.FromSeconds(5);
 157
 158    /// <summary>
 159    /// The hosting shutdown budget that must contain
 160    /// <see cref="KafkaSubscriberOptions.BackgroundDrainTimeout"/> when a subscriber uses
 161    /// <see cref="KafkaAckMode.AckAfterEnqueue"/>.
 162    /// </summary>
 3163    public TimeSpan? HostShutdownTimeout { get; set; } = TimeSpan.FromSeconds(30);
 164
 165    /// <summary>
 166    /// Optional last-chance hook over the producer configuration (compression, linger, batch size,
 167    /// security settings, …) after the package applies its defaults.
 168    /// </summary>
 169    public Action<ProducerConfig>? ConfigureProducer { get; set; }
 170
 171    /// <summary>
 172    /// Optional last-chance hook over each subscriber's consumer configuration (fetch sizes,
 173    /// <c>max.poll.interval.ms</c>, security settings, …) after the package applies its defaults.
 174    /// The package relies on <c>enable.auto.commit=true</c> with <c>enable.auto.offset.store=false</c>
 175    /// for its manual offset management; overriding those breaks delivery guarantees.
 176    /// </summary>
 177    public Action<ConsumerConfig>? ConfigureConsumer { get; set; }
 178
 179    /// <summary>Optional last-chance hook over the admin client configuration used for topic creation.</summary>
 180    public Action<AdminClientConfig>? ConfigureAdminClient { get; set; }
 181
 182    /// <summary>Adds or replaces a named Kafka reply target.</summary>
 183    public KafkaAsyncResponseTransportOptions AddReplyTarget(string name, string responseTopic)
 184    {
 2185        ArgumentException.ThrowIfNullOrWhiteSpace(name);
 2186        ArgumentException.ThrowIfNullOrWhiteSpace(responseTopic);
 187
 2188        ReplyTargets[name] = new KafkaReplyTargetOptions { ResponseTopic = responseTopic };
 3189        return this;
 190    }
 191}
 192
 193/// <summary>Options for one named Kafka async-response reply target.</summary>
 194public sealed class KafkaReplyTargetOptions
 195{
 196    /// <summary>Kafka topic remote systems should produce response payloads to.</summary>
 197    public string? ResponseTopic { get; set; }
 198
 199    /// <summary>Consumer group that receives responses for this target. Optional metadata.</summary>
 200    public string? ConsumerGroup { get; set; }
 201
 202    /// <summary>Additional values copied to the transport-neutral reply target.</summary>
 203    public Dictionary<string, string> Properties { get; } = new(StringComparer.Ordinal);
 204}
Generated by: ReportGenerator 5.5.10.0
08/03/2026 - 08:59:44
GitHub | reportgenerator.io

Methods/Properties

.ctor()
AddReplyTarget(string, string)