fix: stop flow manager after calling revoke handlers

Farfetch · Jul 20, 2024 · e37c8b6 · e37c8b6
1 parent 433f337
commit e37c8b6
Showing 1 changed file with 1 addition and 1 deletion.
diff --git a/src/KafkaFlow/Consumers/Consumer.cs b/src/KafkaFlow/Consumers/Consumer.cs
@@ -242,11 +242,11 @@ private void EnsureConsumer()
             .SetPartitionsRevokedHandler(
                 (consumer, partitions) =>
                 {
+                    _partitionsRevokedHandlers.ForEach(handler => handler(_dependencyResolver, consumer, partitions));
                     this.Assignment = new List<TopicPartition>();
                     this.Subscription = new List<string>();
                     _currentPartitionsOffsets.Clear();
                     _flowManager.Stop();
-                    _partitionsRevokedHandlers.ForEach(handler => handler(_dependencyResolver, consumer, partitions));
                 })
             .SetErrorHandler((consumer, error) => _errorsHandlers.ForEach(x => x(consumer, error)))
             .SetStatisticsHandler((consumer, statistics) => _statisticsHandlers.ForEach(x => x(consumer, statistics)));