Skip to content

Make Kafka Producer and Consumer Factories Thread-Safe #105

Description

@GreenCodeLine

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.

Metadata

Metadata

Assignees

Labels

No labels
No labels

Type

No type

Projects

No projects

Milestone

No milestone

Relationships

None yet

Development

No branches or pull requests

Issue actions