feat: allow use of cluster defined values for topic creation

Farfetch · Apr 10, 2024 · 652b013 · 652b013
1 parent 188b21d
commit 652b013
Show file tree

Hide file tree

Showing 3 changed files with 8 additions and 6 deletions.
diff --git a/src/KafkaFlow.Abstractions/Configuration/IClusterConfigurationBuilder.cs b/src/KafkaFlow.Abstractions/Configuration/IClusterConfigurationBuilder.cs
@@ -75,11 +75,11 @@ public interface IClusterConfigurationBuilder
     /// Adds a Topic to the Cluster
     /// </summary>
     /// <param name="topicName">The topic name</param>
-    /// <param name="numberOfPartitions">The number of Topic partitions</param>
-    /// <param name="replicationFactor">The Topic replication factor</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>
     /// <returns></returns>
     IClusterConfigurationBuilder CreateTopicIfNotExists(
         string topicName,
-        int numberOfPartitions,
-        short replicationFactor);
+        int numberOfPartitions = -1,
+        short replicationFactor = -1);
 }
diff --git a/src/KafkaFlow/Configuration/ClusterConfigurationBuilder.cs b/src/KafkaFlow/Configuration/ClusterConfigurationBuilder.cs
@@ -110,8 +110,8 @@ public IClusterConfigurationBuilder OnStarted(Action<IDependencyResolver> handle
 
     public IClusterConfigurationBuilder CreateTopicIfNotExists(
         string topicName,
-        int numberOfPartitions,
-        short replicationFactor)
+        int numberOfPartitions = -1,
+        short replicationFactor = -1)
     {
         _topicsToCreateIfNotExist.Add(new TopicConfiguration(topicName, numberOfPartitions, replicationFactor));
         return this;

diff --git a/tests/KafkaFlow.IntegrationTests/Core/Bootstrapper.cs b/tests/KafkaFlow.IntegrationTests/Core/Bootstrapper.cs
@@ -42,6 +42,7 @@ internal static class Bootstrapper
     private const string ProtobufGzipTopicName2 = "test-protobuf-gzip-2";
     private const string AvroTopicName = "test-avro";
     private const string NullTopicName = "test-null";
+    private const string DefaultParamsTopicName = "test-default-params";
 
     private static readonly Lazy<IServiceProvider> s_lazyProvider = new(SetupProvider);
 
@@ -202,6 +203,7 @@ private static void SetupServices(HostBuilderContext context, IServiceCollection
                         .CreateTopicIfNotExists(ProtobufGzipTopicName, 2, 1)
                         .CreateTopicIfNotExists(ProtobufGzipTopicName2, 2, 1)
                         .CreateTopicIfNotExists(NullTopicName, 1, 1)
+                        .CreateTopicIfNotExists(DefaultParamsTopicName)
                         .AddConsumer(
                             consumer => consumer
                                 .Topic(ProtobufTopicName)