fix: fix IndexOutOfRangeException on registering offset actions

src/KafkaFlow.Abstractions/Extensions/DictionaryExtensions.cs

@@ -17,7 +17,7 @@ public static class DictionaryExtensions
         /// <typeparam name="TKey">The key type</typeparam>
         /// <typeparam name="TValue">The value type</typeparam>
         /// <returns></returns>
-        public static TValue GetOrAdd<TKey, TValue>(
+        public static TValue SafeGetOrAdd<TKey, TValue>(
             this IDictionary<TKey, TValue> dictionary,
             TKey key,
             Func<TKey, TValue> addFactory)

src/KafkaFlow/Consumers/OffsetManager.cs

@@ -1,13 +1,14 @@
 namespace KafkaFlow.Consumers
 {
     using System;
+    using System.Collections.Concurrent;
     using System.Collections.Generic;
     using System.Linq;
     using Confluent.Kafka;
 
     internal class OffsetManager : IOffsetManager, IDisposable
     {
-        private readonly Dictionary<TopicPartitionOffset, List<Action>> onProcessedActions = new();
+        private readonly ConcurrentDictionary<TopicPartitionOffset, ConcurrentBag<Action>> onProcessedActions = new();
 
         private readonly IOffsetCommitter committer;
         private readonly Dictionary<(string, int), PartitionOffsets> partitionsOffsets;
@@ -46,7 +47,7 @@ public void MarkAsProcessed(TopicPartitionOffset offset)
         public void OnOffsetProcessed(TopicPartitionOffset offset, Action action)
         {
             this.onProcessedActions
-                .GetOrAdd(offset, _ => new List<Action>())
+                .SafeGetOrAdd(offset, _ => new())
                 .Add(action);
         }
 
@@ -72,7 +73,7 @@ private void ExecuteOffsetActions(TopicPartitionOffset offset)
                 return;
             }
 
-            this.onProcessedActions.Remove(offset);
+            this.onProcessedActions.TryRemove(offset, out _);
 
             foreach (var action in actions)
             {

src/KafkaFlow/MiddlewareExecutor.cs

@@ -98,7 +98,7 @@ private IMessageMiddleware GetConsumerOrProducerInstance(
             int index,
             MiddlewareConfiguration configuration)
         {
-            return this.consumerOrProducerMiddlewares.GetOrAdd(
+            return this.consumerOrProducerMiddlewares.SafeGetOrAdd(
                 index,
                 _ => CreateInstance(dependencyResolver, configuration));
         }
@@ -109,7 +109,7 @@ private IMessageMiddleware GetWorkerInstance(
             IMessageContext context,
             MiddlewareConfiguration configuration)
         {
-            return this.workersMiddlewares.GetOrAdd(
+            return this.workersMiddlewares.SafeGetOrAdd(
                 (index, context.ConsumerContext?.WorkerId ?? 0),
                 _ => CreateInstance(dependencyResolver, configuration));
         }