fix: fix NullReferenceEXception when stopping

Farfetch · Jun 28, 2022 · eb8124b · eb8124b
1 parent 6695dd2
commit eb8124b
Showing 1 changed file with 3 additions and 2 deletions.
diff --git a/src/KafkaFlow/Consumers/ConsumerWorkerPool.cs b/src/KafkaFlow/Consumers/ConsumerWorkerPool.cs
@@ -84,16 +84,17 @@ public async Task StopAsync()
 
         public async Task EnqueueAsync(ConsumeResult<byte[], byte[]> message, CancellationToken stopCancellationToken)
         {
+            var localOffsetManager = this.offsetManager;
             var worker = (IConsumerWorker) await this.distributionStrategy
                 .GetWorkerAsync(message.Message.Key, stopCancellationToken)
                 .ConfigureAwait(false);
 
-            if (worker == null)
+            if (worker is null || localOffsetManager is null)
             {
                 return;
             }
 
-            this.offsetManager.Enqueue(message.TopicPartitionOffset);
+            localOffsetManager.Enqueue(message.TopicPartitionOffset);
 
             await worker
                 .EnqueueAsync(message, stopCancellationToken)