Kafka: Update ProduceMessage activity with support for specifying a Key (#6166)

* Add Key to Kafka ProduceMessage activity

Deleted unnecessary Consumer and Producer workflow classes and the OrderReceived message class to clean up code. Refactored Kafka producer interface and implementation to include message keys for improved message handling. Updated configuration to enable Kafka and removed unused service registrations.

* Add Kafka factory classes and type alias registry

Introduce GenericConsumerFactory and GenericProducerFactory for handling Kafka consumer and producer creation. Implement a TypeAliasRegistry to manage type aliases, enabling cleaner configuration through aliases. Update the OrderReceived message class and ensure better integration with the server web program via these new components.

* Handle empty topics and predicates in Kafka worker.

Ensure the Kafka consumer unsubscribes when no topics are available to subscribe to. Additionally, add a check to handle empty string values for predicates, allowing workflow triggers to proceed in this scenario.

* Disable Kafka usage in Elsa Server Web configuration

Kafka has been disabled in the current configuration by setting the useKafka constant to false. This change might be intended to switch to a different messaging system or to simplify the current setup by removing unnecessary services. Ensure that any dependencies on Kafka are handled elsewhere in the application.

Author: sfmskywalker

src/apps/Elsa.Server.Web/Messages/OrderReceived.cs

@@ -3,5 +3,4 @@ namespace Elsa.Server.Web.Messages;
 public class OrderReceived
 {
     public string OrderId { get; set; } = default!;
-    public decimal OrderTotal { get; set; } = default!;
 }

src/apps/Elsa.Server.Web/Program.cs

@@ -4,6 +4,7 @@
 using Elsa.Alterations.MassTransit.Extensions;
 using Elsa.Common.DistributedHosting.DistributedLocks;
 using Elsa.Common.RecurringTasks;
+using Elsa.Common.Serialization;
 using Elsa.Dapper.Extensions;
 using Elsa.Dapper.Services;
 using Elsa.DropIns.Extensions;
@@ -16,6 +17,7 @@
 using Elsa.Features.Services;
 using Elsa.Identity.Multitenancy;
 using Elsa.Kafka;
+using Elsa.Kafka.Factories;
 using Elsa.MassTransit.Extensions;
 using Elsa.MongoDb.Extensions;
 using Elsa.MongoDb.Modules.Alterations;
@@ -30,11 +32,11 @@
 using Elsa.Server.Web.Extensions;
 using Elsa.Server.Web.Filters;
 using Elsa.Server.Web.Messages;
-using Elsa.Server.Web.WorkflowContextProviders;
 using Elsa.Tenants.AspNetCore;
 using Elsa.Tenants.Extensions;
 using Elsa.Workflows.Api;
 using Elsa.Workflows.LogPersistence;
+using Elsa.Workflows.Management;
 using Elsa.Workflows.Management.Compression;
 using Elsa.Workflows.Management.Stores;
 using Elsa.Workflows.Runtime.Distributed.Extensions;
@@ -93,6 +95,10 @@
 var distributedLockProviderName = configuration.GetSection("Runtime:DistributedLocking")["Provider"];
 var appRole = Enum.Parse<ApplicationRole>(configuration["AppRole"] ?? "Default");
 
+// Optionally create type aliases for easier configuration.
+TypeAliasRegistry.RegisterAlias("OrderReceivedProducerFactory", typeof(GenericProducerFactory<string, OrderReceived>));
+TypeAliasRegistry.RegisterAlias("OrderReceivedConsumerFactory", typeof(GenericConsumerFactory<string, OrderReceived>));
+
 // Add Elsa services.
 services
     .AddElsa(elsa =>
@@ -198,6 +204,7 @@
 
                 management.SetDefaultLogPersistenceMode(LogPersistenceMode.Inherit);
                 management.UseReadOnlyMode(useReadOnlyMode);
+                management.AddVariableTypeAndAlias<OrderReceived>("Application");
             })
             .UseProtoActor(proto =>
             {
@@ -426,8 +433,6 @@
                         // etc.
                     });
                 }
-
-                massTransit.AddMessageType<OrderReceived>();
             });
         }
 
@@ -454,8 +459,6 @@
             {
                 kafka.ConfigureOptions(options => configuration.GetSection("Kafka").Bind(options));
             });
-
-            services.AddWorkflowContextProvider<ConsumerDefinitionWorkflowContextProvider>();
         }
 
         if (useAgents)

src/apps/Elsa.Server.Web/WorkflowContextProviders/ConsumerConfigWorkflowContextProvider.cs

@@ -206,21 +206,13 @@
       {
         "Id": "topic-2",
         "Name": "topic-2"
-      },
-      {
-        "Id": "topic-3",
-        "Name": "topic-3"
-      },
-      {
-        "Id": "topic-4",
-        "Name": "topic-4"
       }
     ],
     "Producers": [
       {
         "Id": "producer-1",
         "Name": "Producer 1",
-        "FactoryType": "Elsa.Kafka.Factories.ExpandoObjectProducerFactory, Elsa.Kafka",
+        "FactoryType": "Elsa.Kafka.Factories.GenericProducerFactory`2[[System.String, System.Private.CoreLib], [Elsa.Server.Web.Messages.OrderReceived, Elsa.Server.Web]], Elsa.Kafka",
         "Config": {
           "BootstrapServers": "localhost:9092"
         }
@@ -230,7 +222,7 @@
       {
         "Id": "consumer-1",
         "Name": "Consumer 1",
-        "FactoryType": "Elsa.Kafka.Factories.ExpandoObjectConsumerFactory, Elsa.Kafka",
+        "FactoryType": "Elsa.Kafka.Factories.GenericConsumerFactory`2[[System.String, System.Private.CoreLib], [Elsa.Server.Web.Messages.OrderReceived, Elsa.Server.Web]], Elsa.Kafka",
         "Config": {
           "BootstrapServers": "localhost:9092",
           "GroupId": "group-1",

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

@@ -0,0 +1,10 @@
+namespace Elsa.Common.Serialization;
+
+public static class TypeAliasRegistry
+{
+    public static Dictionary<string, Type> TypeAliases { get; } = new();
+
+    public static void RegisterAlias(string alias, Type type) => TypeAliases[alias] = type;
+
+    public static Type? GetType(string alias) => TypeAliases.GetValueOrDefault(alias);
+}

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

@@ -15,7 +15,11 @@ public override bool CanConvertFrom(ITypeDescriptorContext? context, Type source
     public override object? ConvertFrom(ITypeDescriptorContext? context, CultureInfo? culture, object value)
     {
         if (value is string stringValue)
+        {
+            if (TypeAliasRegistry.GetType(stringValue) is { } type)
+                return type;
             return Type.GetType(stringValue);
+        }
         return base.ConvertFrom(context, culture, value);
     }
 
@@ -27,7 +31,11 @@ public override bool CanConvertTo(ITypeDescriptorContext? context, Type? destina
     public override object? ConvertTo(ITypeDescriptorContext? context, CultureInfo? culture, object? value, Type destinationType)
     {
         if (destinationType == typeof(string) && value is Type type)
+        {
+            if (TypeAliasRegistry.TypeAliases.FirstOrDefault(x => x.Value == type).Key is { } alias)
+                return alias;
             return type.AssemblyQualifiedName;
+        }
         return base.ConvertTo(context, culture, value, destinationType);
     }
 }

src/apps/Elsa.Server.Web/appsettings.json

@@ -7,6 +7,7 @@
 using Elsa.Workflows.Models;
 using Elsa.Workflows.Runtime;
 using Elsa.Workflows.UIHints;
+using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Options;
 
 namespace Elsa.Kafka.Activities;
@@ -48,25 +49,33 @@ public class ProduceMessage : CodeActivity
     public Input<string?> CorrelationId { get; set; } = default!;
 
     /// <summary>
-    /// The content of the message to send.
+    /// The content of the message to produce.
     /// </summary>
     [Input(Description = "The content of the message to produce.")]
     public Input<object> Content { get; set; } = default!;
 
+    /// <summary>
+    /// The key of the message to send.
+    /// </summary>
+    [Input(Description = "The key of the message to produce.")]
+    public Input<object?> Key { get; set; } = default!;
+
     protected override async ValueTask ExecuteAsync(ActivityExecutionContext context)
     {
+        var cancellationToken = context.CancellationToken;
         var topic = Topic.Get(context);
         var producerDefinitionId = ProducerDefinitionId.Get(context);
         var producerDefinitionEnumerator = context.GetRequiredService<IProducerDefinitionEnumerator>();
         var producerDefinition = await producerDefinitionEnumerator.GetByIdAsync(producerDefinitionId);
         var content = Content.Get(context);
+        var key = Key.Get(context);
+
+        if (key is string keyString && string.IsNullOrWhiteSpace(keyString))
+            key = null;
 
-        context.DeferTask(async () =>
-        {
-            using var producer = CreateProducer(context, producerDefinition);
-            var headers = CreateHeaders(context);
-            await producer.ProduceAsync(topic, content, headers);
-        });
+        using var producer = CreateProducer(context, producerDefinition);
+        var headers = CreateHeaders(context);
+        await producer.ProduceAsync(topic, key, content, headers, cancellationToken);
     }
 
     private Headers CreateHeaders(ActivityExecutionContext context)
@@ -84,14 +93,14 @@ private Headers CreateHeaders(ActivityExecutionContext context)
 
         return headers;
     }
-    
+
     private IProducer CreateProducer(ActivityExecutionContext context, ProducerDefinition producerDefinition)
     {
-        var factory = context.GetRequiredService(producerDefinition.FactoryType) as IProducerFactory;
-        
+        var factory = context.GetOrCreateService(producerDefinition.FactoryType) as IProducerFactory;
+
         if (factory == null)
             throw new InvalidOperationException($"Producer factory of type '{producerDefinition.FactoryType}' not found.");
-        
+
         var createProducerContext = new CreateProducerContext(producerDefinition);
         return factory.CreateProducer(createProducerContext);
     }

src/modules/Elsa.Common/Serialization/TypeAliasRegistry.cs

@@ -4,5 +4,5 @@ namespace Elsa.Kafka;
 
 public interface IProducer : IDisposable
 {
-    Task ProduceAsync(string topic, object value, Headers? headers = null, CancellationToken cancellationToken = default);
+    Task ProduceAsync(string topic, object? key, object value, Headers? headers = null, CancellationToken cancellationToken = default);
 }

src/modules/Elsa.Common/Serialization/TypeTypeConverter.cs

@@ -0,0 +1,16 @@
+using Confluent.Kafka;
+using Elsa.Kafka.Implementations;
+using Elsa.Kafka.Serializers;
+
+namespace Elsa.Kafka.Factories;
+
+public class GenericConsumerFactory<TKey, TValue> : IConsumerFactory
+{
+    public IConsumer CreateConsumer(CreateConsumerContext context)
+    {
+        var consumer = new ConsumerBuilder<TKey, TValue>(context.ConsumerDefinition.Config)
+            .SetValueDeserializer(new JsonDeserializer<TValue>())
+            .Build();
+        return new ConsumerProxy(consumer);
+    }
+}

src/modules/Elsa.Kafka/Activities/ProduceMessage.cs

@@ -0,0 +1,16 @@
+using Confluent.Kafka;
+using Elsa.Kafka.Implementations;
+using Elsa.Kafka.Serializers;
+
+namespace Elsa.Kafka.Factories;
+
+public class GenericProducerFactory<TKey, TValue> : IProducerFactory
+{
+    public IProducer CreateProducer(CreateProducerContext workerContext)
+    {
+        var producer = new ProducerBuilder<TKey, TValue>(workerContext.ProducerDefinition.Config)
+            .SetValueSerializer(new JsonSerializer<TValue>())
+            .Build();
+        return new ProducerProxy(producer);
+    }
+}

src/modules/Elsa.Kafka/Contracts/IProducer.cs

@@ -183,6 +183,9 @@ private async Task<bool> EvaluatePredicateAsync(KafkaTransportMessage transportM
 
         if (predicate == null)
             return true;
+        
+        if(string.IsNullOrWhiteSpace(predicate.Value as string))
+            return true;
 
         var expressionExecutionContext = await GetExpressionExecutionContextAsync(transportMessage, binding, cancellationToken);
 

src/modules/Elsa.Kafka/Factories/GenericConsumerFactory.cs

@@ -7,7 +7,7 @@ public class ProducerProxy(object producer) : IProducer
 {
     private object Producer { get; } = producer;
 
-    public async Task ProduceAsync(string topic, object value, Headers? headers = null, CancellationToken cancellationToken = default)
+    public async Task ProduceAsync(string topic, object? key, object value, Headers? headers = null, CancellationToken cancellationToken = default)
     {
         var producerType = Producer.GetType();
         var keyType = producerType.GetGenericArguments()[0];
@@ -16,14 +16,13 @@ public async Task ProduceAsync(string topic, object value, Headers? headers = nu
         var produceAsyncMethod = producerType.GetMethod("ProduceAsync", [typeof(string), messageType, typeof(CancellationToken)])!;
         var messageInstance = Activator.CreateInstance(messageType);
         var convertedValue = value.ConvertTo(valueType);
-        
+
         messageType.GetProperty("Value")!.SetValue(messageInstance, convertedValue);
-        
-        if (headers != null)
-            messageType.GetProperty("Headers")!.SetValue(messageInstance, headers);
-        
+        if (key != null) messageType.GetProperty("Key")!.SetValue(messageInstance, key);
+        if (headers != null) messageType.GetProperty("Headers")!.SetValue(messageInstance, headers);
+
         await (Task)produceAsyncMethod.Invoke(Producer, [topic, messageInstance, cancellationToken])!;
-        
+
         var flushMethod = producerType.GetMethod("Flush", [typeof(CancellationToken)])!;
         flushMethod.Invoke(Producer, [cancellationToken]);
     }

src/modules/Elsa.Kafka/Factories/GenericProducerFactory.cs

@@ -90,7 +90,10 @@ private void Subscribe(IEnumerable<string> topics)
             return;
 
         _subscribedTopics = topicList.ToHashSet();
-        consumer.Subscribe(_subscribedTopics);
+        if(_subscribedTopics.Any())
+            consumer.Subscribe(_subscribedTopics);
+        else
+            consumer.Unsubscribe();
 
         logger.LogInformation("Subscribed to topics: {Topics}", string.Join(", ", _subscribedTopics));
     }

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

@@ -165,8 +165,9 @@ public void StopWorkers()
     private IWorker CreateWorker(IServiceProvider serviceProvider, ConsumerDefinition consumerDefinition)
     {
         var factoryType = consumerDefinition.FactoryType;
-
-        if (serviceProvider.GetRequiredService(factoryType) is not IConsumerFactory consumerFactory)
+        var consumerFactory = ActivatorUtilities.GetServiceOrCreateInstance(serviceProvider, factoryType) as IConsumerFactory;
+        
+        if (consumerFactory == null)
             throw new InvalidOperationException($"Worker factory of type '{factoryType}' not found.");
 
         var createConsumerContext = new CreateConsumerContext(consumerDefinition);