[Host.Kafka] Add configurable client builder factories with logging support

src/SlimMessageBus.Host.Kafka/Configs/KafkaClientConfig.cs

@@ -0,0 +1,9 @@
+namespace SlimMessageBus.Host.Kafka;
+
+public class KafkaClientConfig<T>
+where T : ClientConfig
+{
+    public T ConfluentConfig { get; set; }
+
+    public ILogger Logger { get; set; }
+}

src/SlimMessageBus.Host.Kafka/Configs/KafkaMessageBusSettings.cs

@@ -1,8 +1,8 @@
 namespace SlimMessageBus.Host.Kafka;
-
-using ConsumerBuilder = Confluent.Kafka.ConsumerBuilder<Confluent.Kafka.Ignore, byte[]>;
-using ProducerBuilder = Confluent.Kafka.ProducerBuilder<byte[], byte[]>;
-
+
+using ConsumerBuilder = Confluent.Kafka.ConsumerBuilder<Confluent.Kafka.Ignore, byte[]>;
+using ProducerBuilder = Confluent.Kafka.ProducerBuilder<byte[], byte[]>;
+
 public class KafkaMessageBusSettings
 {
     /// <summary>
@@ -12,7 +12,7 @@ public class KafkaMessageBusSettings
     /// <summary>
     /// Factory method that creates a <see cref="ProducerBuilder"/> based on the supplied settings.
     /// </summary>
-    public Func<ProducerConfig, ProducerBuilder> ProducerBuilderFactory { get; set; }
+    public Func<KafkaClientConfig<ProducerConfig>, ProducerBuilder> ProducerBuilderFactory { get; set; }
     /// <summary>
     /// Factory method that created a <see cref="Producer"/>.
     /// See also: https://kafka.apache.org/documentation/#producerconfigs
@@ -21,7 +21,7 @@ public class KafkaMessageBusSettings
     /// <summary>
     /// Factory method that creates a <see cref="ConsumerBuilder"/> based on the supplied settings and consumer GroupId. 
     /// </summary>
-    public Func<ConsumerConfig, ConsumerBuilder> ConsumerBuilderFactory { get; set; }
+    public Func<KafkaClientConfig<ConsumerConfig>,ConsumerBuilder> ConsumerBuilderFactory { get; set; }
     /// <summary>
     /// Factory method that creates settings based on the consumer GroupId.
     /// See also: https://kafka.apache.org/documentation/#newconsumerconfigs
@@ -32,30 +32,30 @@ public class KafkaMessageBusSettings
     /// </summary>
     public TimeSpan ConsumerPollRetryInterval { get; set; }
 
-    /// <summary>
-    /// Serializer used to serialize Kafka message header values. If not specified the default serializer will be used (setup as part of the bus config). By default the <see cref="DefaultKafkaHeaderSerializer"/> is used.
+    /// <summary>
+    /// Serializer used to serialize Kafka message header values. If not specified the default serializer will be used (setup as part of the bus config). By default the <see cref="DefaultKafkaHeaderSerializer"/> is used.
     /// </summary>
-    public IMessageSerializerProvider HeaderSerializer { get; set; }
-
-    /// <summary>
-    /// Should the commit on partitions for the consumed messages happen when the bus is stopped (or disposed)?
-    /// This ensures the message reprocessing is minimized in between application restarts.
-    /// Default is true.
+    public IMessageSerializerProvider HeaderSerializer { get; set; }
+
+    /// <summary>
+    /// Should the commit on partitions for the consumed messages happen when the bus is stopped (or disposed)?
+    /// This ensures the message reprocessing is minimized in between application restarts.
+    /// Default is true.
     /// </summary>
     public bool EnableCommitOnBusStop { get; set; } = true;
 
     public KafkaMessageBusSettings()
-    {
+    {
         ProducerConfig = (config) => { };
-        ProducerBuilderFactory = (config) => new ProducerBuilder<byte[], byte[]>(config);
+        ProducerBuilderFactory = (config) => new ProducerBuilder<byte[], byte[]>(config.ConfluentConfig);
         ConsumerConfig = (config) => { };
-        ConsumerBuilderFactory = (config) => new ConsumerBuilder<Ignore, byte[]>(config);
-        ConsumerPollRetryInterval = TimeSpan.FromSeconds(2);
+        ConsumerBuilderFactory = (config) => new ConsumerBuilder<Ignore, byte[]>(config.ConfluentConfig);
+        ConsumerPollRetryInterval = TimeSpan.FromSeconds(2);
         HeaderSerializer = new DefaultKafkaHeaderSerializer();
-    }
-
+    }
+
     public KafkaMessageBusSettings(string brokerList) : this()
-    {
+    {
         BrokerList = brokerList;
     }
 }

src/SlimMessageBus.Host.Kafka/Consumer/KafkaGroupConsumer.cs

@@ -68,17 +68,22 @@
 
     protected IConsumer CreateConsumer(string group)
     {
-        var config = new ConsumerConfig
+        var config = new KafkaClientConfig<ConsumerConfig>
         {
-            GroupId = group,
-            BootstrapServers = ProviderSettings.BrokerList
+            ConfluentConfig = new ConsumerConfig
+            {
+                GroupId = group,
+                BootstrapServers = ProviderSettings.BrokerList
+            },
+            Logger = Logger
         };
-        ProviderSettings.ConsumerConfig(config);
+
+        ProviderSettings.ConsumerConfig(config.ConfluentConfig);
 
         // ToDo: add support for auto commit
-        config.EnableAutoCommit = false;
+        config.ConfluentConfig.EnableAutoCommit = false;
         // Notify when we reach EoF, so that we can do a manual commit
-        config.EnablePartitionEof = true;
+        config.ConfluentConfig.EnablePartitionEof = true;
 
         var consumer = ProviderSettings.ConsumerBuilderFactory(config)
             .SetStatisticsHandler((_, json) => OnStatistics(json))
@@ -141,7 +146,7 @@
                 }
             }
             catch (OperationCanceledException)
             {
             }
 
             Logger.LogInformation("Group [{Group}]: Unsubscribing from topics", Group);
@@ -267,7 +272,7 @@
 
     protected virtual void OnStatistics(string json)
     {
         Logger.LogTrace("Group [{Group}]: Statistics: {statistics}", Group, json);
     }
 
     #region Implementation of IKafkaCommitController

src/SlimMessageBus.Host.Kafka/KafkaMessageBus.cs

@@ -45,13 +45,17 @@
     public IProducer CreateProducerInternal()
     {
         _logger.LogTrace("Creating producer settings");
-        var config = new ProducerConfig
+        var config = new KafkaClientConfig<ProducerConfig>
         {
-            BootstrapServers = ProviderSettings.BrokerList,
+            ConfluentConfig = new ProducerConfig()
+            {
+                BootstrapServers = ProviderSettings.BrokerList   
+            },
+            Logger =  _logger
         };
-        ProviderSettings.ProducerConfig(config);
+        ProviderSettings.ProducerConfig(config.ConfluentConfig);
 
-        _producerDeliveryReportsEnabled = config.EnableDeliveryReports ?? true; // when not set per Confluent.KafkaNet docs, it's defaulting to true
+        _producerDeliveryReportsEnabled = config.ConfluentConfig.EnableDeliveryReports ?? true; // when not set per Confluent.KafkaNet docs, it's defaulting to true
 
         _logger.LogDebug("Producer settings: {ProducerSettings}", config);
         return ProviderSettings.ProducerBuilderFactory(config).Build();
@@ -135,7 +139,7 @@
         }
     }
 
     public override async Task ProduceToTransport(object message, Type messageType, string path, IDictionary<string, object> messageHeaders, IMessageBusTarget targetBus, CancellationToken cancellationToken)
     {
         static void OnMessageDelivered(ILogger logger, DeliveryResult<byte[], byte[]> deliveryResult, object message, Type messageType)
         {

src/Tests/SlimMessageBus.Host.Kafka.Test/Consumer/KafkaGroupLoopConsumerTests.cs

@@ -39,9 +39,9 @@ public KafkaLoopGroupConsumerTests()
             .Setup(builder => builder.Build()) 
             .Returns(mockKafkaConsumer.Object);
 
-        var mockConsumerBuilderFactory = new Mock<Func<ConsumerConfig, ConsumerBuilder<Ignore, byte[]>>>(); 
+        var mockConsumerBuilderFactory = new Mock<Func<KafkaClientConfig<ConsumerConfig>, ConsumerBuilder<Ignore, byte[]>>>(); 
         mockConsumerBuilderFactory
-            .Setup(factory => factory(It.IsAny<ConsumerConfig>()))
+            .Setup(factory => factory(It.IsAny<KafkaClientConfig<ConsumerConfig>>()))
             .Returns(mockConsumerBuilder.Object);
 
         var settings = new KafkaMessageBusSettings