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

• Both factories are safe for concurrent calls.
• No direct use of non-thread-safe Dictionary remains in either factory.
• ContainsKey followed by Add is removed.
• ConcurrentDictionary.GetOrAdd is used for producer and consumer creation.
• KafkaProducerFactory continues to reuse the shared producer engine.
• KafkaConsumerFactory preserves the existing Kafka configuration and topic subscription behavior.
• Existing interfaces and public method signatures remain unchanged.
• Existing functionality and disposal behavior are preserved.