Make Kafka Producer and Consumer Factories Thread-Safe
Description
Update the Kafka Producer and Consumer Factory implementations to be thread-safe and safe for concurrent access.
The current implementations use Dictionary<TKey, TValue> together with ContainsKey/Add, which can cause race conditions when multiple threads call Create() concurrently.
The factories should be updated to use ConcurrentDictionary and atomic GetOrAdd operations.
Scope
KafkaProducerFactory
- Replace
Dictionary<string, KafkaProducer<T>> with ConcurrentDictionary<string, KafkaProducer<T>>.
- Update the
Create(string topic) method to use GetOrAdd.
- Ensure that the shared Kafka producer engine remains reused across topics.
- Keep the existing public API and method signatures unchanged.
KafkaConsumerFactory
-
Replace Dictionary<string, KafkaConsumer<T>> with ConcurrentDictionary<string, KafkaConsumer<T>>.
-
Update the Create(string topic, string groupId) method to use GetOrAdd.
-
Preserve the existing consumer configuration, including:
BootstrapServers
AutoOffsetReset.Earliest
GroupId
receive.message.max.bytes
-
Keep the existing public API and method signatures unchanged.
Acceptance Criteria
Make Kafka Producer and Consumer Factories Thread-Safe
Description
Update the Kafka Producer and Consumer Factory implementations to be thread-safe and safe for concurrent access.
The current implementations use
Dictionary<TKey, TValue>together withContainsKey/Add, which can cause race conditions when multiple threads callCreate()concurrently.The factories should be updated to use
ConcurrentDictionaryand atomicGetOrAddoperations.Scope
KafkaProducerFactory
Dictionary<string, KafkaProducer<T>>withConcurrentDictionary<string, KafkaProducer<T>>.Create(string topic)method to useGetOrAdd.KafkaConsumerFactory
Replace
Dictionary<string, KafkaConsumer<T>>withConcurrentDictionary<string, KafkaConsumer<T>>.Update the
Create(string topic, string groupId)method to useGetOrAdd.Preserve the existing consumer configuration, including:
BootstrapServersAutoOffsetReset.EarliestGroupIdreceive.message.max.bytesKeep the existing public API and method signatures unchanged.
Acceptance Criteria
Dictionaryremains in either factory.ContainsKeyfollowed byAddis removed.ConcurrentDictionary.GetOrAddis used for producer and consumer creation.KafkaProducerFactorycontinues to reuse the shared producer engine.KafkaConsumerFactorypreserves the existing Kafka configuration and topic subscription behavior.