| | | 1 | | namespace AsyncResponse.Transports.Kafka; |
| | | 2 | | |
| | | 3 | | internal static class KafkaTransportOptionsValidator |
| | | 4 | | { |
| | | 5 | | /// <summary>Validates the supplied options.</summary> |
| | | 6 | | public static string Required(string? value, string name) |
| | 13360 | 7 | | => !string.IsNullOrWhiteSpace(value) |
| | 13360 | 8 | | ? value |
| | 13360 | 9 | | : throw new InvalidOperationException($"{nameof(KafkaAsyncResponseTransportOptions)}.{name} must be configur |
| | | 10 | | |
| | | 11 | | /// <summary> |
| | | 12 | | /// librdkafka's allowed range for its interval settings (<c>auto.commit.interval.ms</c>, |
| | | 13 | | /// <c>max.poll.interval.ms</c>) tops out at 86,400,000 ms; a larger value fails consumer |
| | | 14 | | /// CONSTRUCTION inside the subscriber loop, not validation. |
| | | 15 | | /// </summary> |
| | 6 | 16 | | private static readonly TimeSpan MaxLibrdkafkaInterval = TimeSpan.FromDays(1); |
| | | 17 | | |
| | | 18 | | /// <summary>Validates a subscriber's <c>max.poll.interval.ms</c> against the librdkafka range.</summary> |
| | | 19 | | internal static void EnsureMaxPollInterval(TimeSpan value, string optionsName, string name) |
| | | 20 | | { |
| | 982 | 21 | | if (value <= TimeSpan.Zero || value > MaxLibrdkafkaInterval) |
| | 6 | 22 | | throw new InvalidOperationException( |
| | 6 | 23 | | $"{optionsName}.{name} must be positive and at most 1 day " + |
| | 6 | 24 | | "(the librdkafka max.poll.interval.ms range; larger values fail consumer construction)."); |
| | 976 | 25 | | } |
| | | 26 | | |
| | | 27 | | /// <summary> |
| | | 28 | | /// The Kafka client passes timeouts to librdkafka as 32-bit millisecond values |
| | | 29 | | /// (<c>Flush(TimeSpan)</c>, <c>Consume(TimeSpan)</c>, admin request timeouts); anything above |
| | | 30 | | /// int.MaxValue ms (~24.8 days) overflows there mid-operation. |
| | | 31 | | /// </summary> |
| | | 32 | | internal static void EnsureIntMilliseconds(TimeSpan value, string optionsName, string name) |
| | | 33 | | { |
| | 3676 | 34 | | if (value <= TimeSpan.Zero || value.TotalMilliseconds > int.MaxValue) |
| | 10 | 35 | | throw new InvalidOperationException( |
| | 10 | 36 | | $"{optionsName}.{name} must be positive and at most {TimeSpan.FromMilliseconds(int.MaxValue).TotalDays:0 |
| | 10 | 37 | | "(the Kafka client passes it to librdkafka as a 32-bit millisecond value)."); |
| | 3666 | 38 | | } |
| | | 39 | | |
| | | 40 | | private static void EnsureDeadLetterNotLive(string deadLetterTopic, KafkaTransportTopicSchema topics) |
| | | 41 | | { |
| | 3372 | 42 | | if (StringComparer.Ordinal.Equals(deadLetterTopic, topics.WorkerTopic) || |
| | 3372 | 43 | | StringComparer.Ordinal.Equals(deadLetterTopic, topics.ResponseTopic)) |
| | | 44 | | { |
| | 4 | 45 | | throw new InvalidOperationException( |
| | 4 | 46 | | $"{nameof(KafkaAsyncResponseTransportOptions)}.{nameof(KafkaAsyncResponseTransportOptions.DeadLetterTopi |
| | 4 | 47 | | $"'{deadLetterTopic}', which is a live transport topic; dead-lettered messages republished into a consum |
| | 4 | 48 | | "are re-consumed instead of parked."); |
| | | 49 | | } |
| | 3368 | 50 | | } |
| | | 51 | | |
| | | 52 | | /// <summary>Validates the supplied options.</summary> |
| | | 53 | | public static void ValidateCommon(KafkaAsyncResponseTransportOptions options) |
| | | 54 | | { |
| | 1714 | 55 | | _ = Required(options.BootstrapServers, nameof(options.BootstrapServers)); |
| | 1708 | 56 | | _ = Required(options.TopicPrefix, nameof(options.TopicPrefix)); |
| | 1706 | 57 | | _ = Required(options.WorkerConsumerGroup, nameof(options.WorkerConsumerGroup)); |
| | 1704 | 58 | | _ = Required(options.ResponseConsumerGroup, nameof(options.ResponseConsumerGroup)); |
| | 1702 | 59 | | _ = Required(options.CorrelationIdHeader, nameof(options.CorrelationIdHeader)); |
| | 1700 | 60 | | _ = Required(options.DefaultReplyTargetName, nameof(options.DefaultReplyTargetName)); |
| | | 61 | | |
| | 1698 | 62 | | if (options.DeadLetterEnabled && string.IsNullOrWhiteSpace(options.DeadLetterTopic)) |
| | 1688 | 63 | | _ = Required(options.DeadLetterTopicSuffix, nameof(options.DeadLetterTopicSuffix)); |
| | | 64 | | |
| | | 65 | | // Worker and response subscribers run distinct consumer groups, so a shared topic feeds |
| | | 66 | | // every message to BOTH: job envelopes complete real waiters through the response ingress |
| | | 67 | | // (the correlation-id header is stamped and trusted) while responses churn through the |
| | | 68 | | // worker dispatcher into the dead-letter topic. Compare the RESOLVED names, as the |
| | | 69 | | // queue-based siblings do, so an explicit value colliding with the other role's derived |
| | | 70 | | // default is caught too. |
| | 1696 | 71 | | var topics = new KafkaTransportTopicSchema(options); |
| | 1696 | 72 | | if (StringComparer.Ordinal.Equals(topics.WorkerTopic, topics.ResponseTopic)) |
| | | 73 | | { |
| | 4 | 74 | | throw new InvalidOperationException( |
| | 4 | 75 | | $"{nameof(KafkaAsyncResponseTransportOptions)}.{nameof(options.WorkerTopic)} and " + |
| | 4 | 76 | | $"{nameof(KafkaAsyncResponseTransportOptions)}.{nameof(options.ResponseTopic)} must resolve to distinct |
| | 4 | 77 | | $"(both resolve to '{topics.WorkerTopic}') so worker and response subscribers do not consume each other' |
| | | 78 | | } |
| | | 79 | | |
| | 1692 | 80 | | if (options.DeadLetterEnabled) |
| | | 81 | | { |
| | | 82 | | // A dead-letter topic aimed at a LIVE transport topic re-feeds poison to a subscriber: |
| | | 83 | | // its own source topic loops the message forever, the other role's topic cross-routes |
| | | 84 | | // it. Derived names ({source}{suffix}) cannot collide with their source, but an |
| | | 85 | | // explicit DeadLetterTopic can. |
| | 1688 | 86 | | EnsureDeadLetterNotLive(topics.DeadLetterTopicFor(topics.WorkerTopic), topics); |
| | 1684 | 87 | | EnsureDeadLetterNotLive(topics.DeadLetterTopicFor(topics.ResponseTopic), topics); |
| | | 88 | | } |
| | | 89 | | |
| | 1688 | 90 | | if (options.OffsetCommitInterval <= TimeSpan.Zero || options.OffsetCommitInterval > MaxLibrdkafkaInterval) |
| | 4 | 91 | | throw new InvalidOperationException( |
| | 4 | 92 | | $"{nameof(KafkaAsyncResponseTransportOptions)}.{nameof(options.OffsetCommitInterval)} must be positive a |
| | 4 | 93 | | "(the librdkafka auto.commit.interval.ms range; larger values fail consumer construction)."); |
| | 1684 | 94 | | EnsureIntMilliseconds(options.OperationTimeout, nameof(KafkaAsyncResponseTransportOptions), nameof(options.Opera |
| | 1680 | 95 | | AsyncResponseChannelOptions.EnsureTimerBacked(options.PublishRetryBaseDelay, nameof(KafkaAsyncResponseTransportO |
| | 1678 | 96 | | AsyncResponseChannelOptions.EnsureTimerBacked(options.PublishRetryMaxDelay, nameof(KafkaAsyncResponseTransportOp |
| | 1678 | 97 | | AsyncResponseChannelOptions.EnsureTimerBacked(options.SubscriberRetryBaseDelay, nameof(KafkaAsyncResponseTranspo |
| | 1678 | 98 | | AsyncResponseChannelOptions.EnsureTimerBacked(options.SubscriberRetryMaxDelay, nameof(KafkaAsyncResponseTranspor |
| | | 99 | | |
| | 1678 | 100 | | if (options.PublishMaxAttempts <= 0) |
| | 2 | 101 | | throw new InvalidOperationException($"{nameof(KafkaAsyncResponseTransportOptions)}.{nameof(options.PublishMa |
| | | 102 | | |
| | 1676 | 103 | | if (options.PublishRetryBaseDelay > options.PublishRetryMaxDelay) |
| | | 104 | | { |
| | 2 | 105 | | throw new InvalidOperationException( |
| | 2 | 106 | | $"{nameof(KafkaAsyncResponseTransportOptions)}.{nameof(options.PublishRetryBaseDelay)} cannot exceed " + |
| | 2 | 107 | | $"{nameof(KafkaAsyncResponseTransportOptions)}.{nameof(options.PublishRetryMaxDelay)}."); |
| | | 108 | | } |
| | | 109 | | |
| | 1674 | 110 | | if (options.SubscriberRetryBaseDelay > options.SubscriberRetryMaxDelay) |
| | | 111 | | { |
| | 2 | 112 | | throw new InvalidOperationException( |
| | 2 | 113 | | $"{nameof(KafkaAsyncResponseTransportOptions)}.{nameof(options.SubscriberRetryBaseDelay)} cannot exceed |
| | 2 | 114 | | $"{nameof(KafkaAsyncResponseTransportOptions)}.{nameof(options.SubscriberRetryMaxDelay)}."); |
| | | 115 | | } |
| | | 116 | | |
| | 1672 | 117 | | if (options.TopicNumPartitions is not -1 and <= 0) |
| | 4 | 118 | | throw new InvalidOperationException($"{nameof(KafkaAsyncResponseTransportOptions)}.{nameof(options.TopicNumP |
| | | 119 | | |
| | 1668 | 120 | | if (options.TopicReplicationFactor is not (-1) and <= 0) |
| | 2 | 121 | | throw new InvalidOperationException($"{nameof(KafkaAsyncResponseTransportOptions)}.{nameof(options.TopicRepl |
| | | 122 | | |
| | 1666 | 123 | | if (options.HostShutdownTimeout is { } hostShutdownTimeout && hostShutdownTimeout <= TimeSpan.Zero) |
| | 2 | 124 | | throw new InvalidOperationException($"{nameof(KafkaAsyncResponseTransportOptions)}.{nameof(options.HostShutd |
| | 1664 | 125 | | } |
| | | 126 | | } |