Add support for per-broker and per-topic partition metrics

Sources/Kafka/Configuration/KafkaConsumerConfig.swift

@@ -31,8 +31,14 @@ public struct KafkaConsumerConfig: Sendable {
     /// The strategy used for consuming messages.
     public var consumptionStrategy: KafkaConsumerConfiguration.ConsumptionStrategy?
 
-    /// Consumer metrics configuration.
-    public var metrics: KafkaConfiguration.ConsumerMetrics = .init()
+    /// Metrics configuration. Set to a ``KafkaMetrics`` instance to enable
+    /// automatic publishing of librdkafka statistics as swift-metrics gauges.
+    /// Requires ``statisticsIntervalMs`` to be configured.
+    public var metrics: KafkaMetrics?
+
+    internal var metricsEnabled: Bool {
+        ((self.statisticsIntervalMs ?? 0) > 0) && (self.metrics != nil)
+    }
 
     // MARK: - Properties generated from librdkafka config list
 
@@ -1178,13 +1184,6 @@ public struct KafkaConsumerConfig: Sendable {
             }
         }
 
-        if config["statistics.interval.ms"] == nil,
-            metrics.enabled,
-            let updateInterval = metrics.updateInterval
-        {
-            config["statistics.interval.ms"] = String(updateInterval.inMilliseconds)
-        }
-
         return config
     }
 

Sources/Kafka/Configuration/KafkaConsumerConfig.swift.gyb

@@ -29,8 +29,14 @@ public struct KafkaConsumerConfig: Sendable {
     /// The strategy used for consuming messages.
     public var consumptionStrategy: KafkaConsumerConfiguration.ConsumptionStrategy?
 
-    /// Consumer metrics configuration.
-    public var metrics: KafkaConfiguration.ConsumerMetrics = .init()
+    /// Metrics configuration. Set to a ``KafkaMetrics`` instance to enable
+    /// automatic publishing of librdkafka statistics as swift-metrics gauges.
+    /// Requires ``statisticsIntervalMs`` to be configured.
+    public var metrics: KafkaMetrics?
+
+    internal var metricsEnabled: Bool {
+      ((self.statisticsIntervalMs ?? 0) > 0) && (self.metrics != nil)
+    }
 
     // MARK: - Properties generated from librdkafka config list
 
@@ -91,13 +97,6 @@ public struct KafkaConsumerConfig: Sendable {
             }
         }
 
-        if config["statistics.interval.ms"] == nil,
-            metrics.enabled,
-            let updateInterval = metrics.updateInterval
-        {
-            config["statistics.interval.ms"] = String(updateInterval.inMilliseconds)
-        }
-
         return config
     }
 

Sources/Kafka/Configuration/KafkaConsumerConfiguration.swift

@@ -222,9 +222,6 @@ public struct KafkaConsumerConfiguration {
     /// Reconnect options.
     public var reconnect: KafkaConfiguration.ReconnectOptions = .init()
 
-    /// Options for librdkafka metrics updates
-    public var metrics: KafkaConfiguration.ConsumerMetrics = .init()
-
     /// Security protocol to use (plaintext, ssl, sasl_plaintext, sasl_ssl).
     /// Default: `.plaintext`
     public var securityProtocol: KafkaConfiguration.SecurityProtocol = .plaintext
@@ -296,12 +293,6 @@ extension KafkaConsumerConfiguration {
         resultDict["reconnect.backoff.ms"] = String(reconnect.backoff.rawValue)
         resultDict["reconnect.backoff.max.ms"] = String(reconnect.maximumBackoff.inMilliseconds)
 
-        if self.metrics.enabled,
-            let updateInterval = self.metrics.updateInterval
-        {
-            resultDict["statistics.interval.ms"] = String(updateInterval.inMilliseconds)
-        }
-
         // Merge with SecurityProtocol configuration dictionary
         resultDict.merge(securityProtocol.dictionary) { _, _ in
             fatalError("securityProtocol and \(#file) should not have duplicate keys")
@@ -321,7 +312,6 @@ extension KafkaConsumerConfiguration {
         // Non-rdkafka properties
         config.pollInterval = self.pollInterval
         config.consumptionStrategy = self.consumptionStrategy
-        config.metrics = self.metrics
 
         return config
     }

Sources/Kafka/Configuration/KafkaProducerConfig.swift

@@ -36,8 +36,14 @@ public struct KafkaProducerConfig: Sendable {
     /// Default: `10000`
     public var shutdownFlushTimeoutMs: Int = 10000
 
-    /// Producer metrics configuration.
-    public var metrics: KafkaConfiguration.ProducerMetrics = .init()
+    /// Metrics configuration. Set to a ``KafkaMetrics`` instance to enable
+    /// automatic publishing of librdkafka statistics as swift-metrics gauges.
+    /// Requires ``statisticsIntervalMs`` to be configured.
+    public var metrics: KafkaMetrics?
+
+    internal var metricsEnabled: Bool {
+        ((self.statisticsIntervalMs ?? 0) > 0) && (self.metrics != nil)
+    }
 
     // MARK: - Properties generated from librdkafka config list
 
@@ -1147,13 +1153,6 @@ public struct KafkaProducerConfig: Sendable {
         config["partitioner"] = self.partitioner?.description
         config["compression.level"] = self.compressionLevel?.description
 
-        if config["statistics.interval.ms"] == nil,
-            metrics.enabled,
-            let updateInterval = metrics.updateInterval
-        {
-            config["statistics.interval.ms"] = String(updateInterval.inMilliseconds)
-        }
-
         return config
     }
 

Sources/Kafka/Configuration/KafkaProducerConfig.swift.gyb

@@ -34,8 +34,14 @@ public struct KafkaProducerConfig: Sendable {
     /// Default: `10000`
     public var shutdownFlushTimeoutMs: Int = 10000
 
-    /// Producer metrics configuration.
-    public var metrics: KafkaConfiguration.ProducerMetrics = .init()
+    /// Metrics configuration. Set to a ``KafkaMetrics`` instance to enable
+    /// automatic publishing of librdkafka statistics as swift-metrics gauges.
+    /// Requires ``statisticsIntervalMs`` to be configured.
+    public var metrics: KafkaMetrics?
+
+    internal var metricsEnabled: Bool {
+      ((self.statisticsIntervalMs ?? 0) > 0) && (self.metrics != nil)
+    }
 
     // MARK: - Properties generated from librdkafka config list
 
@@ -79,13 +85,6 @@ public struct KafkaProducerConfig: Sendable {
         % end
       % end
 
-        if config["statistics.interval.ms"] == nil,
-            metrics.enabled,
-            let updateInterval = metrics.updateInterval
-        {
-            config["statistics.interval.ms"] = String(updateInterval.inMilliseconds)
-        }
-
         return config
     }
 

Sources/Kafka/Configuration/KafkaProducerConfiguration.swift

@@ -161,9 +161,6 @@ public struct KafkaProducerConfiguration {
     /// Reconnect options.
     public var reconnect: KafkaConfiguration.ReconnectOptions = .init()
 
-    /// Options for librdkafka metrics updates
-    public var metrics: KafkaConfiguration.ProducerMetrics = .init()
-
     /// Security protocol to use (plaintext, ssl, sasl_plaintext, sasl_ssl).
     /// Default: `.plaintext`
     public var securityProtocol: KafkaConfiguration.SecurityProtocol = .plaintext
@@ -218,12 +215,6 @@ extension KafkaProducerConfiguration {
         resultDict["reconnect.backoff.ms"] = String(self.reconnect.backoff.rawValue)
         resultDict["reconnect.backoff.max.ms"] = String(self.reconnect.maximumBackoff.inMilliseconds)
 
-        if self.metrics.enabled,
-            let updateInterval = self.metrics.updateInterval
-        {
-            resultDict["statistics.interval.ms"] = String(updateInterval.inMilliseconds)
-        }
-
         // Merge with SecurityProtocol configuration dictionary
         resultDict.merge(self.securityProtocol.dictionary) { _, _ in
             fatalError("securityProtocol and \(#file) should not have duplicate keys")
@@ -247,7 +238,6 @@ extension KafkaProducerConfiguration {
         // Non-rdkafka properties
         config.pollInterval = self.pollInterval
         config.shutdownFlushTimeoutMs = self.flushTimeoutMilliseconds
-        config.metrics = self.metrics
 
         return config
     }

Sources/Kafka/KafkaConsumer.swift

@@ -202,7 +202,7 @@ public final class KafkaConsumer: Sendable, Service {
         if !isAutoCommitEnabled {
             subscribedEvents.append(.offsetCommit)
         }
-        if config.metrics.enabled {
+        if config.metricsEnabled {
             subscribedEvents.append(.statistics)
         }
 
@@ -245,7 +245,7 @@ public final class KafkaConsumer: Sendable, Service {
         if !isAutoCommitEnabled {
             subscribedEvents.append(.offsetCommit)
         }
-        if config.metrics.enabled {
+        if config.metricsEnabled {
             subscribedEvents.append(.statistics)
         }
 
@@ -378,7 +378,7 @@ public final class KafkaConsumer: Sendable, Service {
                 for event in events {
                     switch event {
                     case .statistics(let statistics):
-                        self.config.metrics.update(with: statistics)
+                        self.config.metrics?.update(with: statistics)
                     default:
                         break
                     }