From 652b013eb6a9b5c7b4ba8c26be21cc837a49d202 Mon Sep 17 00:00:00 2001
From: Robert Coltheart <robert.coltheart@gmail.com>
Date: Thu, 11 Apr 2024 08:38:12 +1000
Subject: [PATCH] feat: allow use of cluster defined values for topic creation

---
 .../Configuration/IClusterConfigurationBuilder.cs         | 8 ++++----
 .../Configuration/ClusterConfigurationBuilder.cs          | 4 ++--
 tests/KafkaFlow.IntegrationTests/Core/Bootstrapper.cs     | 2 ++
 3 files changed, 8 insertions(+), 6 deletions(-)

diff --git a/src/KafkaFlow.Abstractions/Configuration/IClusterConfigurationBuilder.cs b/src/KafkaFlow.Abstractions/Configuration/IClusterConfigurationBuilder.cs
index 494e0d301..2c6968baa 100644
--- 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
index 6585032b3..cd85129ef 100644
--- 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
index 272df2b19..5e317cc8e 100644
--- 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)