fix: prevent hanging if stop called from multiple threads

Farfetch · Apr 10, 2024 · 40aa4ba · 40aa4ba
1 parent 188b21d
commit 40aa4ba
Showing 1 changed file with 16 additions and 4 deletions.
diff --git a/src/KafkaFlow/KafkaBus.cs b/src/KafkaFlow/KafkaBus.cs
@@ -18,6 +18,8 @@ internal class KafkaBus : IKafkaBus
 
     private readonly List<IConsumerManager> _consumerManagers = new();
 
+    private bool _stopped;
+
     public KafkaBus(
         IDependencyResolver dependencyResolver,
         KafkaConfiguration configuration,
@@ -73,12 +75,22 @@ public async Task StartAsync(CancellationToken stopCancellationToken = default)
 
     public Task StopAsync()
     {
-        foreach (var cluster in _configuration.Clusters)
+        lock (_consumerManagers)
         {
-            cluster.OnStoppingHandler(_dependencyResolver);
-        }
+            if (_stopped)
+            {
+                return Task.CompletedTask;
+            }
 
-        return Task.WhenAll(_consumerManagers.Select(x => x.StopAsync()));
+            _stopped = true;
+
+            foreach (var cluster in _configuration.Clusters)
+            {
+                cluster.OnStoppingHandler(_dependencyResolver);
+            }
+
+            return Task.WhenAll(_consumerManagers.Select(x => x.StopAsync()));
+        }
     }
 
     private async Task CreateMissingClusterTopics(ClusterConfiguration cluster)