Merge Fix/kafka factories concurrent requests into master

Source/BSN.Commons/Infrastructure/Kafka/KafkaConsumerFactory.cs

@@ -1,6 +1,6 @@
 using Confluent.Kafka;
-using Microsoft.Extensions.Options;
-using System.Collections.Generic;
+using System;
+using System.Collections.Concurrent;
 
 namespace BSN.Commons.Infrastructure.Kafka
 {
@@ -11,41 +11,45 @@ public class KafkaConsumerFactory<T> : IKafkaConsumerFactory<T>
         /// <param name="options">Default Options for KafkaConsumers</param>
         public KafkaConsumerFactory(IKafkaConsumerOptions options)
         {
-            _defaultConsumerOptions = options;
-            _consumers = new Dictionary<string, KafkaConsumer<T>>();
+            _defaultConsumerOptions = options ?? throw new ArgumentNullException(nameof(options));
+            _consumers = new ConcurrentDictionary<string, KafkaConsumer<T>>();
         }
 
         /// <inheritdoc/>
         public IKafkaConsumer<T> Create(string topic, string groupId)
         {
-            string consumerKey = topic + ":" + groupId;
+            if (string.IsNullOrWhiteSpace(topic))
+                throw new ArgumentException("Topic cannot be null or empty.", nameof(topic));
 
-            if (_consumers.ContainsKey(consumerKey))
-            {
-                return _consumers[consumerKey];
-            }
+            if (string.IsNullOrWhiteSpace(groupId))
+                throw new ArgumentException("GroupId cannot be null or empty.", nameof(groupId));
 
-            var config = new ConsumerConfig() 
+            var consumerKey = $"{topic}:{groupId}";
+
+            return _consumers.GetOrAdd(consumerKey, _ =>
             {
-                BootstrapServers = _defaultConsumerOptions.BootstrapServers,
-                AutoOffsetReset = AutoOffsetReset.Earliest,
-                GroupId = groupId
-            };
-
-            // Here we did this because the ReceiveMessageMaxBytes in ProducerConfig type
-            // is int and can not accept high values that we expect
-            config.Set("receive.message.max.bytes", _defaultConsumerOptions.ReceiveMessageMaxBytes);
-
-            // Here Null means that the key in kafka message is null
-            // it helps equal distribution of messages in the kafka cluster
-            var consumerEngine = new ConsumerBuilder<Null, T>(config).Build();
-            consumerEngine.Subscribe(topic);
-
-            var consumer = new KafkaConsumer<T>(consumerEngine);
-
-            _consumers.Add(consumerKey, consumer);
-
-            return consumer;
+                var config = new ConsumerConfig
+                {
+                    BootstrapServers = _defaultConsumerOptions.BootstrapServers,
+                    AutoOffsetReset = AutoOffsetReset.Earliest,
+                    GroupId = groupId
+                };
+
+
+                // Here we did this because the ReceiveMessageMaxBytes in ProducerConfig type
+                // is int and can not accept high values that we expect
+                config.Set(
+                    "receive.message.max.bytes",
+                    _defaultConsumerOptions.ReceiveMessageMaxBytes);
+
+                // Here Null means that the key in kafka message is null
+                // it helps equal distribution of messages in the kafka cluster
+                var consumerEngine = new ConsumerBuilder<Null, T>(config).Build();
+
+                consumerEngine.Subscribe(topic);
+
+                return new KafkaConsumer<T>(consumerEngine);
+            });
         }
 
         /// <inheritdoc />
@@ -55,9 +59,10 @@ public void Dispose()
             {
                 consumer.Value.Dispose();
             }
+            _consumers.Clear();
         }
 
-        private readonly Dictionary<string, KafkaConsumer<T>> _consumers;
+        private readonly ConcurrentDictionary<string, KafkaConsumer<T>> _consumers;
         private readonly IKafkaConsumerOptions _defaultConsumerOptions;
     }
 }

Source/BSN.Commons/Infrastructure/Kafka/KafkaProducerFactory.cs

@@ -1,5 +1,5 @@
-using System.Collections.Generic;
-using Confluent.Kafka;
+using Confluent.Kafka;
+using System.Collections.Concurrent;
 
 namespace BSN.Commons.Infrastructure.Kafka
 {
@@ -14,33 +14,25 @@ public KafkaProducerFactory(IKafkaProducerOptions options)
             {
                 BootstrapServers = options.BootstrapServers,
             };
-            
+
             _sharedProducerEngine = new ProducerBuilder<Null, T>(producerConfig).Build();
-            _producers = new Dictionary<string, KafkaProducer<T>>();
+            _producers = new ConcurrentDictionary<string, KafkaProducer<T>>();
         }
 
         /// <inheritdoc />
         public IKafkaProducer<T> Create(string topic)
         {
-            if (_producers.ContainsKey(topic))
-            {
-                return _producers[topic];
-            }
-
-            var producer = new KafkaProducer<T>(_sharedProducerEngine, topic);
-
-            _producers.Add(topic, producer);
-
-            return producer;
+            return _producers.GetOrAdd(topic, t => new KafkaProducer<T>(_sharedProducerEngine, t));
         }
 
         /// <inheritdoc />
-        public void Dispose()
+        public void Dispose() 
         {
             _sharedProducerEngine?.Dispose();
+            _producers.Clear();
         }
 
         private readonly IProducer<Null, T> _sharedProducerEngine;
-        private readonly Dictionary<string, KafkaProducer<T>> _producers;
+        private readonly ConcurrentDictionary<string, KafkaProducer<T>> _producers;
     }
 }