These changes are a quickfix to prevent an infinite loop while dispos…

…ing clients. It does not fix the original problem but helps ELSA to continue working. (#3901)

Co-authored-by: Yannick Laubscher <yannick.laubscher@swissteach.ch>

src/activities/Elsa.Activities.Kafka/Helpers/Consumer.cs

@@ -22,7 +22,8 @@ public static class Consumer
                                 consumer.Subscribe(topic);
                                 while (!cancellationToken.IsCancellationRequested)
                                 {
-                                    var consumeResult = consumer.Consume(cancellationToken);
+                                    // This code gets executed sometimes, even though the consumer has been disposed. This results in an "handle is destroyed" exepction
+                                    var consumeResult = consumer.Consume(cancellationToken); 
                                     if (consumeResult.IsPartitionEOF) continue;
                                     consumer.Commit(consumeResult);
                                     observer.OnNext(consumeResult.Message);

src/activities/Elsa.Activities.Kafka/Services/Client.cs

@@ -67,8 +67,11 @@ public async Task Dispose()
             {
                 try
                 {
-                    _consumer.Unsubscribe();
-                    _consumer.Close();
+                    using (_consumer)
+                    {
+                        _consumer.Unsubscribe();
+                        _consumer.Close();
+                    }                  
                 }
                 catch (Exception e)
                 {

src/activities/Elsa.Activities.Kafka/Services/Worker.cs

@@ -77,8 +77,11 @@ private async Task OnMessageReceivedAsync(KafkaMessageEvent ev)
 
         private async Task OnErrorAsync(Exception e)
         {
-            _logger.LogError("Error on consuming");
-            await _disposeReceiverAction(this, _client);
+            _logger.LogError($"Error on consuming: {e.Message}");
+            if (!(e.GetType() == typeof(ObjectDisposedException))) // This line prevents an infinite loop when unpublishing a KafkaMessageReceived Trigger. Probably a SemaphoreSlim problem.
+            {
+                await _disposeReceiverAction(this, _client);
+            }
         }
 
         private async Task TriggerWorkflowsAsync(KafkaMessageEvent ev, CancellationToken cancellationToken)

src/activities/Elsa.Activities.Kafka/Services/WorkerManager.cs

@@ -150,8 +150,8 @@ private async Task RemoveAndRespawnWorkerAsync(Worker worker, IClient c, string
         private async Task RemoveWorkerAsync(Worker worker)
         {
             _logger.LogDebug("Disposing worker for {QueueOrTopic}", worker.Topic);
-            await worker.DisposeAsync();
             _workers.Remove(worker);
+            await worker.DisposeAsync();
         }
 
         private IEnumerable<Bookmark> Filter(IEnumerable<Bookmark> triggers)