feat: Create topic with additional configuration if present

Makefile

@@ -3,12 +3,12 @@
 init_broker:	
 	@echo command | date
 	@echo Initializing Kafka broker
-	docker-compose -f docker-compose.yml up -d
+	docker compose -f docker-compose.yml up -d
 
 shutdown_broker:
 	@echo command | date
 	@echo Shutting down kafka broker
-	docker-compose -f docker-compose.yml down
+	docker compose -f docker-compose.yml down
 
 restore:
 	dotnet restore KafkaFlow.sln

src/KafkaFlow.Abstractions/Configuration/IClusterConfigurationBuilder.cs

@@ -77,9 +77,11 @@ public interface IClusterConfigurationBuilder
     /// <param name="topicName">The topic name</param>
     /// <param name="numberOfPartitions">The number of Topic partitions. Default is to use the cluster-defined partitions.</param>
     /// <param name="replicationFactor">The Topic replication factor. Default is to use the cluster-defined replication factor.</param>
+    /// <param name="configs">Additional topic creation configuration values.</param>
     /// <returns></returns>
     IClusterConfigurationBuilder CreateTopicIfNotExists(
         string topicName,
         int numberOfPartitions = -1,
-        short replicationFactor = -1);
+        short replicationFactor = -1,
+        Dictionary<string, string> configs = null);
 }

src/KafkaFlow/Clusters/ClusterManager.cs

@@ -122,6 +122,7 @@ public async Task CreateIfNotExistsAsync(IEnumerable<TopicConfiguration> configu
                         Name = topicConfiguration.Name,
                         ReplicationFactor = topicConfiguration.Replicas,
                         NumPartitions = topicConfiguration.Partitions,
+                        Configs = topicConfiguration.Configs,
                     })
                 .ToArray();
 

src/KafkaFlow/Configuration/ClusterConfigurationBuilder.cs

@@ -111,9 +111,10 @@ public IClusterConfigurationBuilder OnStarted(Action<IDependencyResolver> handle
     public IClusterConfigurationBuilder CreateTopicIfNotExists(
         string topicName,
         int numberOfPartitions = -1,
-        short replicationFactor = -1)
+        short replicationFactor = -1,
+        Dictionary<string, string> configs = null)
     {
-        _topicsToCreateIfNotExist.Add(new TopicConfiguration(topicName, numberOfPartitions, replicationFactor));
+        _topicsToCreateIfNotExist.Add(new TopicConfiguration(topicName, numberOfPartitions, replicationFactor, configs));
         return this;
     }
 }

src/KafkaFlow/Configuration/TopicConfiguration.cs

@@ -1,3 +1,5 @@
+using System.Collections.Generic;
+
 namespace KafkaFlow.Configuration;
 
 /// <summary>
@@ -11,11 +13,13 @@ public class TopicConfiguration
     /// <param name="name">The topic name</param>
     /// <param name="partitions">The number of partitions for the topic</param>
     /// <param name="replicas">Replication factor for the topic</param>
-    public TopicConfiguration(string name, int partitions, short replicas)
+    /// <param name="configs">Additional topic creation configuration values.</param>
+    public TopicConfiguration(string name, int partitions, short replicas, Dictionary<string, string> configs)
     {
         this.Name = name;
         this.Partitions = partitions;
         this.Replicas = replicas;
+        this.Configs = configs;
     }
 
     /// <summary>
@@ -32,4 +36,9 @@ public TopicConfiguration(string name, int partitions, short replicas)
     /// Gets the Topic Replication Factor
     /// </summary>
     public short Replicas { get; }
+
+    /// <summary>
+    /// Gets the topic creation configuration
+    /// </summary>
+    public Dictionary<string, string> Configs { get; }
 }