diff --git a/src/SlimMessageBus.Host.Kafka/Configs/KafkaClientConfig.cs b/src/SlimMessageBus.Host.Kafka/Configs/KafkaClientConfig.cs new file mode 100644 index 00000000..c67c388a --- /dev/null +++ b/src/SlimMessageBus.Host.Kafka/Configs/KafkaClientConfig.cs @@ -0,0 +1,9 @@ +namespace SlimMessageBus.Host.Kafka; + +public class KafkaClientConfig +where T : ClientConfig +{ + public T ConfluentConfig { get; set; } + + public ILogger Logger { get; set; } +} \ No newline at end of file diff --git a/src/SlimMessageBus.Host.Kafka/Configs/KafkaMessageBusSettings.cs b/src/SlimMessageBus.Host.Kafka/Configs/KafkaMessageBusSettings.cs index e0d76853..61892cbc 100644 --- a/src/SlimMessageBus.Host.Kafka/Configs/KafkaMessageBusSettings.cs +++ b/src/SlimMessageBus.Host.Kafka/Configs/KafkaMessageBusSettings.cs @@ -1,8 +1,8 @@ namespace SlimMessageBus.Host.Kafka; - -using ConsumerBuilder = Confluent.Kafka.ConsumerBuilder; -using ProducerBuilder = Confluent.Kafka.ProducerBuilder; - + +using ConsumerBuilder = Confluent.Kafka.ConsumerBuilder; +using ProducerBuilder = Confluent.Kafka.ProducerBuilder; + public class KafkaMessageBusSettings { /// @@ -12,7 +12,7 @@ public class KafkaMessageBusSettings /// /// Factory method that creates a based on the supplied settings. /// - public Func ProducerBuilderFactory { get; set; } + public Func, ProducerBuilder> ProducerBuilderFactory { get; set; } /// /// Factory method that created a . /// See also: https://kafka.apache.org/documentation/#producerconfigs @@ -21,7 +21,7 @@ public class KafkaMessageBusSettings /// /// Factory method that creates a based on the supplied settings and consumer GroupId. /// - public Func ConsumerBuilderFactory { get; set; } + public Func,ConsumerBuilder> ConsumerBuilderFactory { get; set; } /// /// 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 /// public TimeSpan ConsumerPollRetryInterval { get; set; } - /// - /// 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 is used. + /// + /// 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 is used. /// - public IMessageSerializerProvider HeaderSerializer { get; set; } - - /// - /// 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; } + + /// + /// 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 bool EnableCommitOnBusStop { get; set; } = true; public KafkaMessageBusSettings() - { + { ProducerConfig = (config) => { }; - ProducerBuilderFactory = (config) => new ProducerBuilder(config); + ProducerBuilderFactory = (config) => new ProducerBuilder(config.ConfluentConfig); ConsumerConfig = (config) => { }; - ConsumerBuilderFactory = (config) => new ConsumerBuilder(config); - ConsumerPollRetryInterval = TimeSpan.FromSeconds(2); + ConsumerBuilderFactory = (config) => new ConsumerBuilder(config.ConfluentConfig); + ConsumerPollRetryInterval = TimeSpan.FromSeconds(2); HeaderSerializer = new DefaultKafkaHeaderSerializer(); - } - + } + public KafkaMessageBusSettings(string brokerList) : this() - { + { BrokerList = brokerList; } } \ No newline at end of file diff --git a/src/SlimMessageBus.Host.Kafka/Consumer/KafkaGroupConsumer.cs b/src/SlimMessageBus.Host.Kafka/Consumer/KafkaGroupConsumer.cs index a77372e0..132f5076 100644 --- a/src/SlimMessageBus.Host.Kafka/Consumer/KafkaGroupConsumer.cs +++ b/src/SlimMessageBus.Host.Kafka/Consumer/KafkaGroupConsumer.cs @@ -68,17 +68,22 @@ protected override async ValueTask DisposeAsyncCore() protected IConsumer CreateConsumer(string group) { - var config = new ConsumerConfig + var config = new KafkaClientConfig { - 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)) diff --git a/src/SlimMessageBus.Host.Kafka/KafkaMessageBus.cs b/src/SlimMessageBus.Host.Kafka/KafkaMessageBus.cs index b218f40f..aae76403 100644 --- a/src/SlimMessageBus.Host.Kafka/KafkaMessageBus.cs +++ b/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 { - 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(); diff --git a/src/Tests/SlimMessageBus.Host.Kafka.Test/Consumer/KafkaGroupLoopConsumerTests.cs b/src/Tests/SlimMessageBus.Host.Kafka.Test/Consumer/KafkaGroupLoopConsumerTests.cs index 95ca5742..6f1f9e5d 100644 --- a/src/Tests/SlimMessageBus.Host.Kafka.Test/Consumer/KafkaGroupLoopConsumerTests.cs +++ b/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>>(); + var mockConsumerBuilderFactory = new Mock, ConsumerBuilder>>(); mockConsumerBuilderFactory - .Setup(factory => factory(It.IsAny())) + .Setup(factory => factory(It.IsAny>())) .Returns(mockConsumerBuilder.Object); var settings = new KafkaMessageBusSettings