Add bounded backpressure for producer and consumer event sequences

Sources/Kafka/Configuration/KafkaConsumerConfig.swift

@@ -34,6 +34,24 @@ public struct KafkaConsumerConfig: Sendable {
     /// Consumer metrics configuration.
     public var metrics: KafkaConfiguration.ConsumerMetrics = .init()
 
+    /// Low watermark for the ``KafkaConsumerEvents`` asynchronous sequence backpressure.
+    /// When the number of buffered events falls below this threshold, the consumer
+    /// will resume yielding events to the sequence.
+    /// Default: `100`
+    public var eventsBackpressureLowWatermark: Int = 100
+
+    /// High watermark for the ``KafkaConsumerEvents`` asynchronous sequence backpressure.
+    /// When the number of buffered events reaches this threshold, the consumer
+    /// will temporarily stop yielding events to the sequence to prevent out-of-memory errors.
+    /// Default: `1000`
+    public var eventsBackpressureHighWatermark: Int = 1000
+
+    /// Arbitrary configuration properties passed directly to librdkafka.
+    ///
+    /// - Warning: Properties set here override typed properties.
+    /// Intended for testing or advanced configurations not explicitly supported by this library.
+    internal var additionalConfig: [String: String] = [:]
+
     // MARK: - Properties generated from librdkafka config list
 
     /// Client identifier.
@@ -1046,14 +1064,6 @@ public struct KafkaConsumerConfig: Sendable {
     /// librdkafka property name: `"consume.callback.max.messages"`
     public var consumeCallbackMaxMessages: Int?
 
-    /// Additional librdkafka configuration properties not covered by typed properties.
-    /// Keys and values are passed directly to librdkafka.
-    ///
-    /// - Warning: Properties set here override typed properties above.
-    /// Intended for testing (e.g. `test.mock.num.brokers`) or advanced configurations
-    /// not explicitly supported by this library.
-    internal var additionalConfig: [String: String] = [:]
-
     public init() {}
 
     internal var config: [String: String] {
@@ -1193,9 +1203,7 @@ public struct KafkaConsumerConfig: Sendable {
             config["statistics.interval.ms"] = String(updateInterval.inMilliseconds)
         }
 
-        for (key, value) in self.additionalConfig {
-            config[key] = value
-        }
+        config.merge(self.additionalConfig) { _, new in new }
 
         return config
     }

Sources/Kafka/Configuration/KafkaConsumerConfig.swift.gyb

@@ -32,6 +32,24 @@ public struct KafkaConsumerConfig: Sendable {
     /// Consumer metrics configuration.
     public var metrics: KafkaConfiguration.ConsumerMetrics = .init()
 
+    /// Low watermark for the ``KafkaConsumerEvents`` asynchronous sequence backpressure.
+    /// When the number of buffered events falls below this threshold, the consumer
+    /// will resume yielding events to the sequence.
+    /// Default: `100`
+    public var eventsBackpressureLowWatermark: Int = 100
+
+    /// High watermark for the ``KafkaConsumerEvents`` asynchronous sequence backpressure.
+    /// When the number of buffered events reaches this threshold, the consumer
+    /// will temporarily stop yielding events to the sequence to prevent out-of-memory errors.
+    /// Default: `1000`
+    public var eventsBackpressureHighWatermark: Int = 1000
+
+    /// Arbitrary configuration properties passed directly to librdkafka.
+    ///
+    /// - Warning: Properties set here override typed properties.
+    /// Intended for testing or advanced configurations not explicitly supported by this library.
+    internal var additionalConfig: [String: String] = [:]
+
     // MARK: - Properties generated from librdkafka config list
 
   % for rdconf in consumer_librdkafka_configs:
@@ -54,7 +72,7 @@ public struct KafkaConsumerConfig: Sendable {
     ///
     /// Default: ${rdconf["defaultValue"]}
     % end
-    public var ${rdconf["swiftName"]}: ${rdconf["swiftType"].replace("enum", f"KafkaConfig.{rdconf["swiftEnumName"]}")}?
+    public var ${rdconf["swiftName"]}: ${rdconf["swiftType"].replace("enum", f"KafkaConfig.{rdconf['swiftEnumName']}")}?
   % end
 
     public init() {}
@@ -98,6 +116,8 @@ public struct KafkaConsumerConfig: Sendable {
             config["statistics.interval.ms"] = String(updateInterval.inMilliseconds)
         }
 
+        config.merge(self.additionalConfig) { _, new in new }
+
         return config
     }
 
@@ -115,9 +135,9 @@ public struct KafkaConsumerConfig: Sendable {
         % elif rdconf["swiftType"] == "[String]":
         self.${rdconf["swiftName"]} = configDict["${rdconf["name"]}"]?.components(separatedBy: ",")${" // ignore-unacceptable-language" if rdconf["name"] == "topic.blacklist" else ""}
         % elif rdconf["swiftType"] == "enum":
-        self.${rdconf["swiftName"]} = configDict["${rdconf["name"]}"].map(KafkaConfig.${rdconf["swiftEnumName"]}.init)
+        self.${rdconf["swiftName"]} = configDict["${rdconf["name"]}"].map(KafkaConfig.${rdconf['swiftEnumName']}.init)
         % elif rdconf["swiftType"] == "[enum]":
-        self.${rdconf["swiftName"]} = configDict["${rdconf["name"]}"]?.components(separatedBy: ",").map { ${f"KafkaConfig.{rdconf["swiftEnumName"]}"}(description: $0) }
+        self.${rdconf["swiftName"]} = configDict["${rdconf["name"]}"]?.components(separatedBy: ",").map { ${f"KafkaConfig.{rdconf['swiftEnumName']}"}(description: $0) }
         % end
       % end
     }

Sources/Kafka/Configuration/KafkaProducerConfig.swift

@@ -39,6 +39,25 @@ public struct KafkaProducerConfig: Sendable {
     /// Producer metrics configuration.
     public var metrics: KafkaConfiguration.ProducerMetrics = .init()
 
+    /// Low watermark for the ``KafkaProducerEvents`` asynchronous sequence backpressure.
+    /// When the number of buffered events falls below this threshold, the producer
+    /// will resume yielding events to the sequence.
+    /// Default: `1000`
+    public var eventsBackpressureLowWatermark: Int = 1000
+
+    /// High watermark for the ``KafkaProducerEvents`` asynchronous sequence backpressure.
+    /// When the number of buffered events reaches this threshold, the producer
+    /// will temporarily stop yielding events to the sequence to prevent out-of-memory errors.
+    /// Polling of librdkafka continues to process delivery reports for `sendAndAwait`.
+    /// Default: `10000` (corresponds to 1 million messages, matching confluent-kafka-go default)
+    public var eventsBackpressureHighWatermark: Int = 10000
+
+    /// Arbitrary configuration properties passed directly to librdkafka.
+    ///
+    /// - Warning: Properties set here override typed properties.
+    /// Intended for testing or advanced configurations not explicitly supported by this library.
+    internal var additionalConfig: [String: String] = [:]
+
     // MARK: - Properties generated from librdkafka config list
 
     /// Client identifier.
@@ -1035,14 +1054,6 @@ public struct KafkaProducerConfig: Sendable {
     /// Default: -1
     public var compressionLevel: Int?
 
-    /// Additional librdkafka configuration properties not covered by typed properties.
-    /// Keys and values are passed directly to librdkafka.
-    ///
-    /// - Warning: Properties set here override typed properties above.
-    /// Intended for testing (e.g. `test.mock.num.brokers`) or advanced configurations
-    /// not explicitly supported by this library.
-    internal var additionalConfig: [String: String] = [:]
-
     public init() {}
 
     internal var config: [String: String] {
@@ -1162,9 +1173,7 @@ public struct KafkaProducerConfig: Sendable {
             config["statistics.interval.ms"] = String(updateInterval.inMilliseconds)
         }
 
-        for (key, value) in self.additionalConfig {
-            config[key] = value
-        }
+        config.merge(self.additionalConfig) { _, new in new }
 
         return config
     }

Sources/Kafka/Configuration/KafkaProducerConfig.swift.gyb

@@ -37,6 +37,25 @@ public struct KafkaProducerConfig: Sendable {
     /// Producer metrics configuration.
     public var metrics: KafkaConfiguration.ProducerMetrics = .init()
 
+    /// Low watermark for the ``KafkaProducerEvents`` asynchronous sequence backpressure.
+    /// When the number of buffered events falls below this threshold, the producer
+    /// will resume yielding events to the sequence.
+    /// Default: `1000`
+    public var eventsBackpressureLowWatermark: Int = 1000
+
+    /// High watermark for the ``KafkaProducerEvents`` asynchronous sequence backpressure.
+    /// When the number of buffered events reaches this threshold, the producer
+    /// will temporarily stop yielding events to the sequence to prevent out-of-memory errors.
+    /// Polling of librdkafka continues to process delivery reports for `sendAndAwait`.
+    /// Default: `10000` (corresponds to 1 million messages, matching confluent-kafka-go default)
+    public var eventsBackpressureHighWatermark: Int = 10000
+
+    /// Arbitrary configuration properties passed directly to librdkafka.
+    ///
+    /// - Warning: Properties set here override typed properties.
+    /// Intended for testing or advanced configurations not explicitly supported by this library.
+    internal var additionalConfig: [String: String] = [:]
+
     // MARK: - Properties generated from librdkafka config list
 
   % for rdconf in producer_librdkafka_configs:
@@ -59,7 +78,7 @@ public struct KafkaProducerConfig: Sendable {
     ///
     /// Default: ${rdconf["defaultValue"]}
     % end
-    public var ${rdconf["swiftName"]}: ${rdconf["swiftType"].replace("enum", f"KafkaConfig.{rdconf["swiftEnumName"]}")}?
+    public var ${rdconf["swiftName"]}: ${rdconf["swiftType"].replace("enum", f"KafkaConfig.{rdconf['swiftEnumName']}")}?
   % end
 
     public init() {}
@@ -86,6 +105,8 @@ public struct KafkaProducerConfig: Sendable {
             config["statistics.interval.ms"] = String(updateInterval.inMilliseconds)
         }
 
+        config.merge(self.additionalConfig) { _, new in new }
+
         return config
     }
 
@@ -103,9 +124,9 @@ public struct KafkaProducerConfig: Sendable {
         % elif rdconf["swiftType"] == "[String]":
         self.${rdconf["swiftName"]} = configDict["${rdconf["name"]}"]?.components(separatedBy: ",")${" // ignore-unacceptable-language" if rdconf["name"] == "topic.blacklist" else ""}
         % elif rdconf["swiftType"] == "enum":
-        self.${rdconf["swiftName"]} = configDict["${rdconf["name"]}"].map(KafkaConfig.${rdconf["swiftEnumName"]}.init)
+        self.${rdconf["swiftName"]} = configDict["${rdconf["name"]}"].map(KafkaConfig.${rdconf['swiftEnumName']}.init)
         % elif rdconf["swiftType"] == "[enum]":
-        self.${rdconf["swiftName"]} = configDict["${rdconf["name"]}"]?.components(separatedBy: ",").map { ${f"KafkaConfig.{rdconf["swiftEnumName"]}"}(description: $0) }
+        self.${rdconf["swiftName"]} = configDict["${rdconf["name"]}"]?.components(separatedBy: ",").map { ${f"KafkaConfig.{rdconf['swiftEnumName']}"}(description: $0) }
         % end
       % end
     }