Repository: kestra-io/kestra
Issue: Prevent FlowableTask to hit Workers
Labels: bug
backend

For now, executor send FlowableTask to the WorkerTask queue and the worker will change it to RUNNING. 

There is no need to have to hit worker for FlowableTask, so we need to keep it on the executor only for this. 

Also it can have some side effect (WorkerInstance restarted, ...) 

Issue comments:

Pull request review comments:

Patch that resolved the issue:
diff --git a/runner-kafka/src/main/java/org/kestra/runner/kafka/KafkaExecutor.java b/runner-kafka/src/main/java/org/kestra/runner/kafka/KafkaExecutor.java
index 44bde4225f..9c712d0afa 100644
--- a/runner-kafka/src/main/java/org/kestra/runner/kafka/KafkaExecutor.java
+++ b/runner-kafka/src/main/java/org/kestra/runner/kafka/KafkaExecutor.java
@@ -24,6 +24,7 @@
 import org.kestra.core.models.flows.State;
 import org.kestra.core.models.templates.Template;
 import org.kestra.core.models.triggers.Trigger;
+import org.kestra.core.queues.AbstractQueue;
 import org.kestra.core.queues.QueueFactoryInterface;
 import org.kestra.core.queues.QueueInterface;
 import org.kestra.core.runners.*;
@@ -541,7 +542,7 @@ private void handleWorkerTask(KStream<String, ExecutionWithFlow> stream) {
                 Named.as("handleWorkerTask-map")
             );
 
-        KStream<String, WorkerTask> result = branchException(streamEither, "handleWorkerTask")
+        KStream<String, WorkerTask> dedupWorkerTask = branchException(streamEither, "handleWorkerTask")
             .flatMapValues((readOnlyKey, value) -> value)
             .transformValues(
                 () -> new DeduplicationTransformer<>(
@@ -552,20 +553,53 @@ private void handleWorkerTask(KStream<String, ExecutionWithFlow> stream) {
                 Named.as("handleWorkerTask-deduplication-transform"),
                 WORKERTASK_DEDUPLICATION_STATE_STORE_NAME
             )
-            .filter((key, value) -> value != null, Named.as("handleWorkerTask-null-filter"))
-            .map((key, value) -> new KeyValue<>(value.getTaskRun().getId(), value))
+            .filter((key, value) -> value != null, Named.as("handleWorkerTask-null-filter"));
+
+        KStream<String, WorkerTaskResult> resultFlowable = dedupWorkerTask
+            .filter((key, value) -> value.getTask().isFlowable(), Named.as("handleWorkerTask-isFlowable-filter"))
+            .mapValues(
+                (key, value) -> new WorkerTaskResult(value.withTaskRun(value.getTaskRun().withState(State.Type.RUNNING))),
+                Named.as("handleWorkerTask-isFlowableToRunning-mapValues")
+            )
+            .map(
+                (key, value) -> new KeyValue<>(AbstractQueue.key(value), value),
+                Named.as("handleWorkerTask-isFlowable-map")
+            )
             .selectKey(
-                (key, value) -> value.getTaskRun().getId(),
-                Named.as("handleWorkerTask-selectKey")
+                (key, value) -> AbstractQueue.key(value),
+                Named.as("handleWorkerTask-isFlowable-selectKey")
+            );
+
+        KStream<String, WorkerTaskResult> workerTaskResultKStream = logIfEnabled(
+            resultFlowable,
+            (key, value) -> log.debug(
+                "WorkerTaskResult out: {}",
+                value.getTaskRun().toStringState()
+            ),
+            "handleWorkerTask-isFlowableLog"
+        );
+
+        workerTaskResultKStream
+            .to(
+                kafkaAdminService.getTopicName(WorkerTaskResult.class),
+                Produced.with(Serdes.String(), JsonSerde.of(WorkerTaskResult.class))
+            );
+
+        KStream<String, WorkerTask> resultNotFlowable = dedupWorkerTask
+            .filter((key, value) -> !value.getTask().isFlowable(), Named.as("handleWorkerTask-notFlowable-filter"))
+            .map((key, value) -> new KeyValue<>(AbstractQueue.key(value), value), Named.as("handleWorkerTask-notFlowableKeyValue-map"))
+            .selectKey(
+                (key, value) -> AbstractQueue.key(value),
+                Named.as("handleWorkerTask-notFlowableKeyValue-selectKey")
             );
 
         KStream<String, WorkerTask> workerTaskKStream = logIfEnabled(
-            result,
+            resultNotFlowable,
             (key, value) -> log.debug(
                 "WorkerTask out: {}",
                 value.getTaskRun().toStringState()
             ),
-            "handleWorkerTask-log"
+            "handleWorkerTask-notFlowableLog"
         );
 
         workerTaskKStream
diff --git a/runner-kafka/src/main/java/org/kestra/runner/kafka/streams/DeduplicationTransformer.java b/runner-kafka/src/main/java/org/kestra/runner/kafka/streams/DeduplicationTransformer.java
index 7fd5e3536d..a5a5b22ec9 100644
--- a/runner-kafka/src/main/java/org/kestra/runner/kafka/streams/DeduplicationTransformer.java
+++ b/runner-kafka/src/main/java/org/kestra/runner/kafka/streams/DeduplicationTransformer.java
@@ -37,7 +37,7 @@ public V transform(final K key, final V value) {
         SV latestValue = store.get(storeKey);
 
         if (latestValue != null && latestValue.equals(currentValue)) {
-            log.warn("Duplicate value for key '{}', storeKey '{}', value '{}'", key, storeKey, latestValue);
+            log.debug("Duplicate value for key '{}', storeKey '{}', value '{}'", key, storeKey, latestValue);
             return null;
         }
 
diff --git a/runner-kafka/src/main/java/org/kestra/runner/kafka/streams/ExecutionNextsDeduplicationTransformer.java b/runner-kafka/src/main/java/org/kestra/runner/kafka/streams/ExecutionNextsDeduplicationTransformer.java
index 6de1288071..b4e8358668 100644
--- a/runner-kafka/src/main/java/org/kestra/runner/kafka/streams/ExecutionNextsDeduplicationTransformer.java
+++ b/runner-kafka/src/main/java/org/kestra/runner/kafka/streams/ExecutionNextsDeduplicationTransformer.java
@@ -43,7 +43,7 @@ public KafkaExecutor.ExecutionNexts transform(final String key, final KafkaExecu
 
         if (groups.get(true).size() > 0) {
             groups.get(true).forEach(s ->
-                log.warn("Duplicate next taskRun for execution '{}', value '{}'", key, s)
+                log.debug("Duplicate next taskRun for execution '{}', value '{}'", key, s)
             );
 
             return null;
diff --git a/runner-memory/src/main/java/org/kestra/runner/memory/MemoryExecutor.java b/runner-memory/src/main/java/org/kestra/runner/memory/MemoryExecutor.java
index 7789bab953..2887bd3486 100644
--- a/runner-memory/src/main/java/org/kestra/runner/memory/MemoryExecutor.java
+++ b/runner-memory/src/main/java/org/kestra/runner/memory/MemoryExecutor.java
@@ -101,11 +101,23 @@ private void handleExecution(ExecutionState state) {
 
                 Optional<List<WorkerTask>> workerTasks = this.doWorkerTask(execution, flow);
                 if (workerTasks.isPresent()) {
-                    workerTasks.stream()
+                    List<WorkerTask> workerTasksDedup = workerTasks.stream()
                         .flatMap(Collection::stream)
                         .filter(workerTask -> this.deduplicateWorkerTask(execution, workerTask.getTaskRun()))
+                        .collect(Collectors.toList());
+
+                    // WorkerTask not flowable to workerTask
+                    workerTasksDedup
+                        .stream()
+                        .filter(workerTask -> !workerTask.getTask().isFlowable())
                         .forEach(workerTaskQueue::emit);
 
+                    // WorkerTask not flowable to workerTaskResult as Running
+                    workerTasksDedup
+                        .stream()
+                        .filter(workerTask -> workerTask.getTask().isFlowable())
+                        .map(workerTask -> new WorkerTaskResult(workerTask.withTaskRun(workerTask.getTaskRun().withState(State.Type.RUNNING))))
+                        .forEach(workerTaskResultQueue::emit);
                     return;
                 }
 
@@ -224,7 +236,7 @@ private boolean deduplicateWorkerTask(Execution execution, TaskRun taskRun) {
         State.Type current = executionState.workerTaskDeduplication.get(deduplicationKey);
 
         if (current == taskRun.getState().getCurrent()) {
-            log.warn("Duplicate WorkerTask on execution '{}' for taskRun '{}', value '{}, taskId '{}'", execution.getId(), taskRun.getId(), taskRun.getValue(), taskRun.getTaskId());
+            log.debug("Duplicate WorkerTask on execution '{}' for taskRun '{}', value '{}, taskId '{}'", execution.getId(), taskRun.getId(), taskRun.getValue(), taskRun.getTaskId());
             return false;
         } else {
             executionState.workerTaskDeduplication.put(deduplicationKey, taskRun.getState().getCurrent());
@@ -241,7 +253,7 @@ private boolean deduplicateNexts(Execution execution, List<TaskRun> taskRuns) {
                 String deduplicationKey = taskRun.getParentTaskRunId() + "-" + taskRun.getTaskId() + "-" + taskRun.getValue();
 
                 if (executionState.childDeduplication.containsKey(deduplicationKey)) {
-                    log.warn("Duplicate Nexts on execution '{}' with key '{}'", execution.getId(), deduplicationKey);
+                    log.debug("Duplicate Nexts on execution '{}' with key '{}'", execution.getId(), deduplicationKey);
                     return false;
                 } else {
                     executionState.childDeduplication.put(deduplicationKey, taskRun.getId());


Test patch: