Merge pull request #420 from Robban1980/feature/kafka-workflow-event-listener

[Feature] Adding support for Kafka in the Workflow Event Listener

Author: v1r3n

common/src/main/java/com/netflix/conductor/common/run/WorkflowSummaryExtended.java

@@ -0,0 +1,62 @@
+/*
+ * Copyright 2025 Conductor Authors.
+ * <p>
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+package com.netflix.conductor.common.run;
+
+import java.util.Map;
+
+import com.netflix.conductor.annotations.protogen.ProtoField;
+
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+/** Extended version of WorkflowSummary that retains input/output as Map */
+public class WorkflowSummaryExtended extends WorkflowSummary {
+
+    @ProtoField(id = 9) // Ensure Protobuf compatibility
+    @JsonIgnore
+    private Map<String, Object> inputMap;
+
+    @ProtoField(id = 10)
+    @JsonIgnore
+    private Map<String, Object> outputMap;
+
+    public WorkflowSummaryExtended(Workflow workflow) {
+        super(workflow);
+        if (workflow.getInput() != null) {
+            this.inputMap = workflow.getInput();
+        }
+        if (workflow.getOutput() != null) {
+            this.outputMap = workflow.getOutput();
+        }
+    }
+
+    /** New method for JSON serialization */
+    @JsonProperty("input")
+    public Map<String, Object> getInputMap() {
+        return inputMap;
+    }
+
+    /** New method for JSON serialization */
+    @JsonProperty("output")
+    public Map<String, Object> getOutputMap() {
+        return outputMap;
+    }
+
+    public void setInputMap(Map<String, Object> inputMap) {
+        this.inputMap = inputMap;
+    }
+
+    public void setOutputMap(Map<String, Object> outputMap) {
+        this.outputMap = outputMap;
+    }
+}

core/src/main/java/com/netflix/conductor/core/execution/WorkflowExecutorOps.java

@@ -14,9 +14,29 @@
 
 import com.netflix.conductor.model.WorkflowModel;
 
+import com.fasterxml.jackson.annotation.JsonValue;
+
 /** Listener for the completed and terminated workflows */
 public interface WorkflowStatusListener {
 
+    enum WorkflowEventType {
+        STARTED,
+        RERAN,
+        RETRIED,
+        PAUSED,
+        RESUMED,
+        RESTARTED,
+        COMPLETED,
+        TERMINATED,
+        FINALIZED;
+
+        @JsonValue // Ensures correct JSON serialization
+        @Override
+        public String toString() {
+            return name().toLowerCase(); // Convert to lowercase for consistency
+        }
+    }
+
     default void onWorkflowCompletedIfEnabled(WorkflowModel workflow) {
         if (workflow.getWorkflowDefinition().isWorkflowStatusListenerEnabled()) {
             onWorkflowCompleted(workflow);
@@ -35,9 +55,57 @@ default void onWorkflowFinalizedIfEnabled(WorkflowModel workflow) {
         }
     }
 
+    default void onWorkflowStartedIfEnabled(WorkflowModel workflow) {
+        if (workflow.getWorkflowDefinition().isWorkflowStatusListenerEnabled()) {
+            onWorkflowStarted(workflow);
+        }
+    }
+
+    default void onWorkflowRestartedIfEnabled(WorkflowModel workflow) {
+        if (workflow.getWorkflowDefinition().isWorkflowStatusListenerEnabled()) {
+            onWorkflowRestarted(workflow);
+        }
+    }
+
+    default void onWorkflowRerunIfEnabled(WorkflowModel workflow) {
+        if (workflow.getWorkflowDefinition().isWorkflowStatusListenerEnabled()) {
+            onWorkflowRerun(workflow);
+        }
+    }
+
+    default void onWorkflowRetriedIfEnabled(WorkflowModel workflow) {
+        if (workflow.getWorkflowDefinition().isWorkflowStatusListenerEnabled()) {
+            onWorkflowRetried(workflow);
+        }
+    }
+
+    default void onWorkflowPausedIfEnabled(WorkflowModel workflow) {
+        if (workflow.getWorkflowDefinition().isWorkflowStatusListenerEnabled()) {
+            onWorkflowPaused(workflow);
+        }
+    }
+
+    default void onWorkflowResumedIfEnabled(WorkflowModel workflow) {
+        if (workflow.getWorkflowDefinition().isWorkflowStatusListenerEnabled()) {
+            onWorkflowResumed(workflow);
+        }
+    }
+
     void onWorkflowCompleted(WorkflowModel workflow);
 
     void onWorkflowTerminated(WorkflowModel workflow);
 
     default void onWorkflowFinalized(WorkflowModel workflow) {}
+
+    default void onWorkflowStarted(WorkflowModel workflow) {}
+
+    default void onWorkflowRestarted(WorkflowModel workflow) {}
+
+    default void onWorkflowRerun(WorkflowModel workflow) {}
+
+    default void onWorkflowPaused(WorkflowModel workflow) {}
+
+    default void onWorkflowResumed(WorkflowModel workflow) {}
+
+    default void onWorkflowRetried(WorkflowModel workflow) {}
 }

core/src/main/java/com/netflix/conductor/core/listener/WorkflowStatusListener.java

@@ -1,7 +1,8 @@
 # Workflow Event Listeners
 Workflow Event listeners can be configured for the purpose in Conductor:
-1. Remove and/or archive workflows from primary datasource (e.g. Redis) once the workflow reaches a terminal status
-2. Publish a message to a conductor queue as the workflows complete that can be used to trigger other workflows 
+1. Remove and/or archive workflows from primary datasource (e.g. Redis) once the workflow reaches a terminal status.
+2. Publish a message to a conductor queue as the workflows complete that can be used to trigger other workflows.
+3. Publish workflow status changes to Kafka as it moves along its lifecycle.
 
 ## Published Artifacts
 
@@ -47,3 +48,65 @@ conductor.workflow-status-listener.queue-publisher.failureQueue=_callbackFailure
 #Queue for terminal state workflows (success or failed)
 conductor.workflow-status-listener.queue-publisher.finalizeQueue=_callbackFinalizeQueue
 ```
+
+### Kafka Publisher
+Publish a summary of workflow WorkflowSummary to a Kafka topic(s) as a workflow moves through its lifecycle.
+
+This publisher introduced some new events
+- STARTED
+- RERAN
+- RETRIED
+- PAUSED
+- RESUMED
+- RESTARTED
+- COMPLETED (supported by queue_publisher)
+- TERMINATED (supported by queue_publisher)
+- FINALIZED (supported by queue_publisher)
+
+Example of a default configuration:
+
+```properties
+conductor.workflow-status-listener.type=kafka
+
+# Kafka Producer Configurations
+conductor.workflow-status-listener.kafka.producer.bootstrap.servers=kafka:29092
+
+# Kafka Producer Configuration
+conductor.workflow-status-listener.kafka.producer.key.serializer=org.apache.kafka.common.serialization.StringSerializer
+conductor.workflow-status-listener.kafka.producer.value.serializer=org.apache.kafka.common.serialization.StringSerializer
+
+# Reliability Settings
+conductor.workflow-status-listener.kafka.producer.acks=all
+conductor.workflow-status-listener.kafka.producer.enable.idempotence=true
+
+# Retry sending messages if failure
+conductor.workflow-status-listener.kafka.producer.retries=5
+conductor.workflow-status-listener.kafka.producer.retry.backoff.ms=100
+
+# Allow batching (default 0)
+conductor.workflow-status-listener.kafka.producer.linger.ms=10
+conductor.workflow-status-listener.kafka.producer.batch.size=65536
+conductor.workflow-status-listener.kafka.producer.buffer.memory=67108864
+# Reduce network load
+conductor.workflow-status-listener.kafka.producer.compression.type=zstd
+
+# Allow multiple in-flight messages (better throughput)
+conductor.workflow-status-listener.kafka.producer.max.in.flight.requests.per.connection=1
+
+# Default Topic for All Workflow Status Events
+conductor.workflow-status-listener.kafka.default-topic=workflow-status-events
+
+```
+
+For configuration it supports the Kafka Producer clients settings prefixed with `conductor.workflow-status-listener.kafka.producer`.
+
+`conductor.workflow-status-listener.kafka.default-topic`  defines the default topic to use for all events.
+Each event can also have its dedicated topic prefix the proeprty with `conductor.workflow-status-listener.kafka.event-topics.` followed by the event name in lowercase.
+
+Example of using specific topics for the events:
+```properties
+# Custom Topics for Specific Events
+conductor.workflow-status-listener.kafka.event-topics.completed=workflow-completed-events
+conductor.workflow-status-listener.kafka.event-topics.terminated=workflow-terminated-events
+conductor.workflow-status-listener.kafka.event-topics.started=workflow-started-events
+```

workflow-event-listener/README.md

@@ -13,6 +13,10 @@ dependencies {
     implementation "org.apache.commons:commons-lang3:"
     implementation group: 'org.apache.httpcomponents', name: 'httpclient', version: '4.5.14'
     implementation "com.amazonaws:aws-java-sdk-s3:${revAwsSdk}"
+    implementation "org.apache.kafka:kafka-clients:${revKafka}"
+
+    implementation "com.fasterxml.jackson.core:jackson-databind"
+    implementation "com.fasterxml.jackson.core:jackson-core"
 
     compileOnly 'org.springframework.boot:spring-boot-starter'
     compileOnly 'org.springframework.boot:spring-boot-starter-web'
@@ -21,14 +25,15 @@ dependencies {
     testImplementation "org.apache.groovy:groovy-all:${revGroovy}"
     testImplementation "org.spockframework:spock-core:${revSpock}"
     testImplementation "org.spockframework:spock-spring:${revSpock}"
+
     implementation "org.springframework.boot:spring-boot-starter-log4j2"
     testImplementation 'org.springframework.retry:spring-retry'
     testImplementation 'org.springframework.boot:spring-boot-starter-web'
+
     testImplementation "com.netflix.dyno-queues:dyno-queues-redis:${revDynoQueues}"
     testImplementation project(':conductor-test-util').sourceSets.test.output
 
     //In memory
     implementation "org.rarefiedredis.redis:redis-java:${revRarefiedRedis}"
     testImplementation "redis.clients:jedis:${revJedis}"
-
 }