chore(minor): [sc-2373] allow to get metadata from broker (#23)

* chore(minor): [sc-2373] allow to get metadata from broker

* use lazy vars

* use map instead of for-loop

* rename classes

Author: blindspotbounty

Sources/Kafka/KafkaConsumer.swift

@@ -454,6 +454,10 @@ public final class KafkaConsumer: Sendable, Service {
             throw KafkaError.client(reason: err)
         }
     }
+    
+    public func metadata() async throws -> KafkaMetadata {
+        try await client().metadata()
+    }
 
     /// Start the ``KafkaConsumer``.
     ///
@@ -1058,7 +1062,7 @@ extension KafkaConsumer {
             case .uninitialized:
                 fatalError("\(#function) invoked while still in state \(self.state)")
             case .initializing:
-                fatalError("Subscribe to consumer group / assign to topic partition pair before reading messages")
+                self.state = .finished
             case .running(let client, _, let eventSource):
                 self.state = .running(client: client, messagePollLoopState: .finished, eventSource: eventSource)
             case .finishing, .finished:

Sources/Kafka/KafkaMetadata.swift

@@ -0,0 +1,54 @@
+import Crdkafka
+
+public final class KafkaMetadata {
+    private let metadata: UnsafePointer<rd_kafka_metadata>
+    
+    init(metadata: UnsafePointer<rd_kafka_metadata>) {
+        self.metadata = metadata
+    }
+    
+    deinit {
+        rd_kafka_metadata_destroy(metadata)
+    }
+    
+    public private(set) lazy var topics = {
+        (0..<Int(self.metadata.pointee.topic_cnt)).map { KafkaTopicMetadata(metadata: self, topic: self.metadata.pointee.topics[$0]) }
+    }()
+}
+
+// must be a class to allow mutating lazy vars, otherwise require struct copies
+public final class KafkaTopicMetadata {
+    private let metadata: KafkaMetadata // retain metadata
+    private let topic: rd_kafka_metadata_topic
+    
+    init(metadata: KafkaMetadata, topic: rd_kafka_metadata_topic) {
+        self.metadata = metadata
+        self.topic = topic
+    }
+
+    public private(set) lazy var name = {
+        String(cString: self.topic.topic)
+    }()
+    
+    public private(set) lazy var partitions = {
+        (0..<Int(self.topic.partition_cnt)).map { KafkaPartitionMetadata(metadata: self.metadata, partition: topic.partitions[$0]) }
+    }()
+}
+
+public struct KafkaPartitionMetadata {
+    private let metadata: KafkaMetadata // retain metadata
+    private let partition: rd_kafka_metadata_partition
+    
+    init(metadata: KafkaMetadata, partition: rd_kafka_metadata_partition) {
+        self.metadata = metadata
+        self.partition = partition
+    }
+    
+    var id: Int {
+        Int(partition.id)
+    }
+
+    var replicasCount: Int {
+        Int(partition.replica_cnt)
+    }
+}

Sources/Kafka/RDKafka/RDKafkaClient.swift

@@ -806,6 +806,22 @@ final class RDKafkaClient: Sendable {
     func withKafkaHandlePointer<T>(_ body: (OpaquePointer) async throws -> T) async rethrows -> T {
         return try await body(self.kafkaHandle)
     }
+    
+    func metadata() async throws -> KafkaMetadata {
+        let queue = DispatchQueue(label: "com.swift-server.swift-kafka.metadata")
+        return try await withCheckedThrowingContinuation { (continuation: CheckedContinuation<KafkaMetadata, Error>) in
+            queue.async {
+                var metadata: UnsafePointer<rd_kafka_metadata>?
+                let error = rd_kafka_metadata(self.kafkaHandle, 1, nil, &metadata, -1)
+                guard error == RD_KAFKA_RESP_ERR_NO_ERROR,
+                      let metadata else {
+                    continuation.resume(throwing: KafkaError.rdKafkaError(wrapping: error))
+                    return
+                }
+                continuation.resume(returning: KafkaMetadata(metadata: metadata))
+            }
+        }
+    }
 
     func initTransactions(timeout: Duration) async throws {
         let result = await performBlockingCall(queue: gcdQueue) {

Tests/IntegrationTests/KafkaTests.swift

@@ -758,7 +758,6 @@ final class KafkaTests: XCTestCase {
             }
 
             group.addTask {
-                let logger = Logger.kafkaTest
                 var messages = [KafkaConsumerMessage]()
                 for try await record in consumer.messages {
                     guard !record.eof else {
@@ -774,7 +773,52 @@ final class KafkaTests: XCTestCase {
             await consumerServiceGroup.triggerGracefulShutdown()
         }
     }
+    
+    func testMetadata() async throws {
+        let uniqueTopic = try createUniqueTopic(partitions: 7)
+        defer {
+            // delete topic
+            var basicConfig = KafkaConsumerConfiguration(
+                consumptionStrategy: .group(id: "no-group", topics: []),
+                bootstrapBrokerAddresses: [self.bootstrapBrokerAddress]
+            )
+            basicConfig.broker.addressFamily = .v4
 
+            let client = try? RDKafkaClient.makeClient(
+                type: .consumer,
+                configDictionary: basicConfig.dictionary,
+                events: [],
+                logger: .kafkaTest
+            )
+            try? client?._deleteTopic(uniqueTopic, timeout: 10 * 1000)
+        }
+        
+        var consumerConfig = KafkaConsumerConfiguration(
+            consumptionStrategy: .group(
+                id: "test",
+                topics: []
+            ),
+            bootstrapBrokerAddresses: [self.bootstrapBrokerAddress]
+        )
+        consumerConfig.autoOffsetReset = .beginning // Read topic from beginning
+        consumerConfig.broker.addressFamily = .v4
+        consumerConfig.enablePartitionEof = true
+
+        let consumer = try KafkaConsumer(
+            configuration: consumerConfig,
+            logger: .kafkaTest
+        )
+        let metadata = try await consumer.metadata()
+        let topic = metadata.topics.first { topic in
+            topic.name == uniqueTopic
+        }
+        guard let topic else {
+            XCTFail("Topic was not found")
+            return
+        }
+        let partitions = topic.partitions
+        XCTAssertEqual(partitions.count, 7)
+    }
     // MARK: - Helpers
 
     func createUniqueTopic(partitions: Int32 = -1 /* default num for cluster */) throws -> String {