#385 - Fix EOS duplicate handle - Client is fenced

core/Kafka/Internal/StreamsProducer.cs

@@ -58,7 +58,7 @@ public StreamsProducer(
                 case ProcessingGuarantee.AT_LEAST_ONCE:
                     break;
                 case ProcessingGuarantee.EXACTLY_ONCE:
-                    _producerConfig.TransactionalId = $"{config.ApplicationId}-{processId}";
+                    _producerConfig.TransactionalId = $"{config.ApplicationId}-{processId}-{threadId}";
                     break;
                 default:
                     throw new StreamsException($"Guarantee {config.Guarantee} is not supported yet");

core/Metrics/Internal/RocksDBMetricsRecordingTrigger.cs

@@ -1,18 +1,19 @@
+using System.Collections.Concurrent;
 using System.Collections.Generic;
+using System.Runtime.InteropServices;
 using Streamiz.Kafka.Net.Errors;
 
 namespace Streamiz.Kafka.Net.Metrics.Internal
 {
     internal class RocksDbMetricsRecordingTrigger
     {
-        private IDictionary<string, RocksDbMetricsRecorder> MetricsRecorders { get; } 
-            = new Dictionary<string, RocksDbMetricsRecorder>();
+        private ConcurrentDictionary<string, RocksDbMetricsRecorder> MetricsRecorders { get; } = new();
 
         internal void AddMetricsRecorder(RocksDbMetricsRecorder recorder)
         {
             if (!MetricsRecorders.ContainsKey(recorder.Name))
             {
-                MetricsRecorders.Add(recorder.Name, recorder);
+                MetricsRecorders.TryAdd(recorder.Name, recorder);
                 return;
             }
 
@@ -21,7 +22,7 @@ internal void AddMetricsRecorder(RocksDbMetricsRecorder recorder)
         }
 
         internal void RemoveMetricsRecorder(RocksDbMetricsRecorder recorder)
-            => MetricsRecorders.Remove(recorder.Name);
+            => MetricsRecorders.TryRemove(recorder.Name, out RocksDbMetricsRecorder _);
 
         internal void Run(long now)
         {

launcher/sample-stream/Program.cs

@@ -21,11 +21,13 @@ public static async Task Main(string[] args)
                 ApplicationId = $"test-app",
                 BootstrapServers = "localhost:9092",
                 AutoOffsetReset = AutoOffsetReset.Earliest,
+                Guarantee = ProcessingGuarantee.EXACTLY_ONCE,
                 Logger = LoggerFactory.Create((b) =>
                 {
                     b.AddConsole();
                     b.SetMinimumLevel(LogLevel.Information);
-                })
+                }),
+                NumStreamThreads = 2
             };
 
             var t = BuildTopology();