wip

Signed-off-by: a.kiselev <saneks222@yandex.ru>

Author: a.kiselev

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

@@ -64,17 +64,22 @@ protected override async ValueTask DisposeAsyncCore()
 
     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))

src/SlimMessageBus.Host.Kafka/KafkaMessageBus.cs

@@ -45,13 +45,17 @@ public void Flush()
     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();