feat: be able to specify configuration for created topic [sc-14009]

Sources/Kafka/ForTesting/RDKafkaClient+Topic.swift

@@ -21,24 +21,26 @@ extension RDKafkaClient {
     /// Create a topic with a unique name (`UUID`).
     /// Blocks for a maximum of `timeout` milliseconds.
     /// - Parameter partitions: Partitions in topic (default: -1 - default for broker)
+    /// - Parameter config: Topic configuration, if empty broker defaults are used. See https://kafka.apache.org/documentation.html#topicconfigs
     /// - Parameter timeout: Timeout in milliseconds.
     /// - Returns: Name of newly created topic.
     /// - Throws: A ``KafkaError`` if the topic creation failed.
-    public func _createUniqueTopic(partitions: Int32 = -1, timeout: Int32) throws -> String {
+    public func _createUniqueTopic(partitions: Int32 = -1, config: [String: String] = [:], timeout: Int32) throws -> String {
         let uniqueTopicName = UUID().uuidString
 
-        try _createTopic(topicName: uniqueTopicName, partitions: partitions, timeout: timeout)
+        try _createTopic(topicName: uniqueTopicName, partitions: partitions, config: config, timeout: timeout)
 
         return uniqueTopicName
     }
 
     /// Create a topic with specified name
     /// Blocks for a maximum of `timeout` milliseconds.
     /// - Parameter partitions: Partitions in topic (default: -1 - default for broker)
+    /// - Parameter config: Topic configuration, if empty broker defaults are used. See https://kafka.apache.org/documentation.html#topicconfigs
     /// - Parameter timeout: Timeout in milliseconds.
     /// - Returns: Name of newly created topic.
     /// - Throws: A ``KafkaError`` if the topic creation failed.
-    public func _createTopic(topicName: String, partitions: Int32 = -1, timeout: Int32) throws {
+    public func _createTopic(topicName: String, partitions: Int32 = -1, config: [String: String] = [:], timeout: Int32) throws {
         let errorChars = UnsafeMutablePointer<CChar>.allocate(capacity: RDKafkaClient.stringSize)
         defer { errorChars.deallocate() }
 
@@ -54,6 +56,13 @@ extension RDKafkaClient {
         }
         defer { rd_kafka_NewTopic_destroy(newTopic) }
 
+        for (key, value) in config {
+            let resultCode = rd_kafka_NewTopic_set_config(newTopic, key, value)
+            guard resultCode == RD_KAFKA_RESP_ERR_NO_ERROR else {
+                throw KafkaError.rdKafkaError(wrapping: resultCode)
+            }
+        }
+
         try self.withKafkaHandlePointer { kafkaHandle in
             let resultQueue = rd_kafka_queue_new(kafkaHandle)
             defer { rd_kafka_queue_destroy(resultQueue) }