Fix error handling and logging in Kafka components

Enhanced error handling in Worker.cs to catch consume exceptions and log warnings. Added logging to TriggerWorkflows.cs for better traceability and fixed issues with expression evaluation. Also updated predicate in ConsumerWorkflow.cs and set CanStartWorkflow to true.

src/apps/Elsa.Server.Web/Workflows/ConsumerWorkflow.cs

@@ -21,9 +21,9 @@ public class ConsumerWorkflow : WorkflowBase
                 {
                     ConsumerDefinitionId = new("consumer-1"),
                     Topics = new(["topic-1"]),
-                    Predicate = new(JavaScriptExpression.Create("message => message.OrderId == '1'")),
+                    Predicate = new(JavaScriptExpression.Create("getMessage().OrderId == '1'")),
                     Result = new(message),
-                    CanStartWorkflow = false
+                    CanStartWorkflow = true
                 },
                 new WriteLine(c => JsonSerializer.Serialize(message.Get(c)))
             }

src/modules/Elsa.Kafka/Handlers/TriggerWorkflows.cs

@@ -10,6 +10,7 @@ using Elsa.Workflows.Memory;
 using Elsa.Workflows.Runtime;
 using Elsa.Workflows.Runtime.Options;
 using JetBrains.Annotations;
+using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
 
 namespace Elsa.Kafka.Handlers;
@@ -21,7 +22,8 @@ public class TriggerWorkflows(
     ICorrelationStrategy correlationStrategy,
     IExpressionEvaluator expressionEvaluator,
     IOptions<KafkaOptions> options,
-    IServiceProvider serviceProvider) : INotificationHandler<TransportMessageReceived>
+    IServiceProvider serviceProvider,
+    ILogger<TriggerWorkflows> logger) : INotificationHandler<TransportMessageReceived>
 {
     private static readonly string MessageReceivedActivityTypeName = ActivityTypeNameHelper.GenerateTypeName<MessageReceived>();
 
@@ -137,7 +139,7 @@ public class TriggerWorkflows(
 
         var correlationId = GetCorrelationId(transportMessage);
         var workflowInstanceId = GetWorkflowInstanceId(transportMessage);
-        
+
         foreach (var binding in boundBookmarks)
         {
             var stimulus = binding.Stimulus;
@@ -176,13 +178,24 @@ public class TriggerWorkflows(
             return true;
 
         var memory = new MemoryRegister();
-        var messageVariable = new Variable("message", transportMessage);
-        var message = transportMessage;
+        var transportMessageVariable = new Variable("transportMessage", transportMessage);
+        var messageVariable = new Variable("message", transportMessage.Value);
         var expressionExecutionContext = new ExpressionExecutionContext(serviceProvider, memory, cancellationToken: cancellationToken);
-        messageVariable.Set(expressionExecutionContext, message);
-        return await expressionEvaluator.EvaluateAsync<bool>(predicate, expressionExecutionContext);
+
+        transportMessageVariable.Set(expressionExecutionContext, transportMessage);
+        messageVariable.Set(expressionExecutionContext, transportMessage.Value);
+
+        try
+        {
+            return await expressionEvaluator.EvaluateAsync<bool>(predicate, expressionExecutionContext);
+        }
+        catch (Exception e)
+        {
+            logger.LogWarning(e, "An error occurred while evaluating the predicate for stimulus {Stimulus}", stimulus);
+            return false;
+        }
     }
-    
+
     private string? GetWorkflowInstanceId(KafkaTransportMessage transportMessage)
     {
         var key = options.Value.WorkflowInstanceIdHeaderKey;

src/modules/Elsa.Kafka/Implementations/Worker.cs

@@ -85,26 +85,44 @@ public class Worker<TKey, TValue>(WorkerContext workerContext, IConsumer<TKey, T
             throw new InvalidOperationException("The worker is not running.");
 
         var topicList = topics.ToHashSet();
-        
+
         if (topicList.SetEquals(_subscribedTopics))
             return;
-        
+
         _subscribedTopics = topicList.ToHashSet();
         consumer.Subscribe(_subscribedTopics);
-        
+
         logger.LogInformation("Subscribed to topics: {Topics}", string.Join(", ", _subscribedTopics));
     }
 
     private async Task RunAsync(CancellationToken cancellationToken)
     {
+        var consumeExceptionCount = 0;
         while (!cancellationToken.IsCancellationRequested)
         {
-            var consumeResult = consumer.Consume(cancellationToken);
+            try
+            {
+                var consumeResult = consumer.Consume(cancellationToken);
+                consumeExceptionCount = 0;
 
-            if (consumeResult.IsPartitionEOF)
-                continue;
+                if (consumeResult.IsPartitionEOF)
+                    continue;
 
-            await ProcessMessageAsync(consumeResult, cancellationToken);
+                await ProcessMessageAsync(consumeResult, cancellationToken);
+            }
+            catch (ConsumeException e)
+            {
+                logger.LogWarning(e, "Error consuming message.");
+                consumeExceptionCount++;
+
+                if (consumeExceptionCount > 100)
+                    throw new InvalidOperationException("Too many consume exceptions.");
+            }
+            catch (OperationCanceledException)
+            {
+                logger.LogInformation("Consumer was cancelled.");
+                break;
+            }
         }
 
         consumer.Unsubscribe();