Repository: kestra-io/kestra
Issue: move io.kestra.core.tasks.flows.Flow to executor
Labels: bug
backend

This task is buggy : 
- If the worker tasks is resubmit, we recreate a new execution
- If there is many `Flow` tasks at the same time, it will used all the thread on the worker and lead to infinite execution (no more worker, we can start task, we listen for something that would never happen)

We need to move the whole logic to executor that know tasks state and can to this in streaming without any blocking worker

Issue comments:

Pull request review comments:

Patch that resolved the issue:
diff --git a/cli/src/main/resources/application.yml b/cli/src/main/resources/application.yml
index 43f9509472..e9b9011c0a 100644
--- a/cli/src/main/resources/application.yml
+++ b/cli/src/main/resources/application.yml
@@ -118,6 +118,12 @@ kestra:
         properties:
           cleanup.policy: "delete,compact"
 
+      executorworkertaskexecution:
+        cls: io.kestra.core.runners.WorkerTaskExecution
+        name: "${kestra.kafka.defaults.topic-prefix}executor_workertaskexecution"
+        properties:
+          cleanup.policy: "delete,compact"
+
       workertask:
         name: "${kestra.kafka.defaults.topic-prefix}workertask"
         cls: io.kestra.core.runners.WorkerTask
@@ -158,6 +164,7 @@ kestra:
         properties:
           cleanup.policy: "compact"
 
+
   elasticsearch:
     defaults:
       indice-prefix: "kestra_"
diff --git a/core/src/main/java/io/kestra/core/runners/AbstractExecutor.java b/core/src/main/java/io/kestra/core/runners/AbstractExecutor.java
index 93cbdc718a..12797713b2 100644
--- a/core/src/main/java/io/kestra/core/runners/AbstractExecutor.java
+++ b/core/src/main/java/io/kestra/core/runners/AbstractExecutor.java
@@ -22,6 +22,7 @@
 import java.util.stream.Collectors;
 
 import static io.kestra.core.utils.Rethrow.throwFunction;
+import static io.kestra.core.utils.Rethrow.throwPredicate;
 
 @Slf4j
 public abstract class AbstractExecutor implements Runnable, Closeable {
@@ -59,6 +60,9 @@ public Executor process(Executor executor) {
             // search for worker task result
             executor = this.handleChildWorkerCreatedKilling(executor);
             executor = this.handleChildWorkerTaskResult(executor);
+
+            // search for flow task
+            executor = this.handleFlowTask(executor);
         } catch (Exception e) {
             return executor.withException(e, "process");
         }
@@ -285,22 +289,21 @@ private Executor onEnd(Executor executor) {
     }
 
     private Executor handleNext(Executor Executor) {
-        List<TaskRun> nexts = this.saveFlowableOutput(
-            FlowableUtils
-                .resolveSequentialNexts(
-                    Executor.getExecution(),
-                    ResolvedTask.of(Executor.getFlow().getTasks()),
-                    ResolvedTask.of(Executor.getFlow().getErrors())
-                ),
-            Executor,
-            null
-        );
+        List<NextTaskRun> nextTaskRuns = FlowableUtils
+            .resolveSequentialNexts(
+                Executor.getExecution(),
+                ResolvedTask.of(Executor.getFlow().getTasks()),
+                ResolvedTask.of(Executor.getFlow().getErrors())
+            );
 
-        if (nexts.size() == 0) {
+        if (nextTaskRuns.size() == 0) {
             return Executor;
         }
 
-        return Executor.withTaskRun(nexts, "handleNext");
+        return Executor.withTaskRun(
+            this.saveFlowableOutput(nextTaskRuns, Executor, null),
+            "handleNext"
+        );
     }
 
     private Executor handleChildNext(Executor executor) throws InternalException {
@@ -468,6 +471,60 @@ private Executor handleWorkerTask(Executor executor) throws InternalException {
         return executor.withWorkerTasks(workerTasks, "handleWorkerTask");
     }
 
+    private Executor handleFlowTask(final Executor executor) throws Exception {
+        List<WorkerTaskExecution> executions = new ArrayList<>();
+        List<WorkerTaskResult> workerTaskResults = new ArrayList<>();
+
+        boolean haveFlows = executor.getWorkerTasks()
+            .removeIf(throwPredicate(workerTask -> {
+                if (!(workerTask.getTask() instanceof io.kestra.core.tasks.flows.Flow)) {
+                    return false;
+                }
+
+                io.kestra.core.tasks.flows.Flow flowTask = (io.kestra.core.tasks.flows.Flow) workerTask.getTask();
+                RunContext runContext = runContextFactory.of(
+                    executor.getFlow(),
+                    flowTask,
+                    executor.getExecution(),
+                    workerTask.getTaskRun()
+                );
+
+                Execution execution = flowTask.createExecution(runContext);
+
+                WorkerTaskExecution workerTaskExecution = WorkerTaskExecution.builder()
+                    .task(flowTask)
+                    .taskRun(workerTask.getTaskRun())
+                    .runContext(runContext)
+                    .execution(execution)
+                    .build();
+
+                executions.add(workerTaskExecution);
+
+                if (!flowTask.getWait()) {
+                    workerTaskResults.add(flowTask.createWorkerTaskResult(
+                        null,
+                        workerTaskExecution,
+                        null,
+                        execution
+                    ));
+                }
+
+                return true;
+            }));
+
+        if (!haveFlows) {
+            return executor;
+        }
+
+        Executor resultExecutor = executor.withWorkerTaskExecutions(executions, "handleFlowTask");
+
+        if (workerTaskResults.size() > 0) {
+            resultExecutor = executor.withWorkerTaskResults(workerTaskResults, "handleFlowTaskWorkerTaskResults");
+        }
+
+        return resultExecutor;
+    }
+
     protected static void log(Logger log, Boolean in, WorkerTask value) {
         log.debug(
             "{} {} : {}",
@@ -486,6 +543,16 @@ protected static void log(Logger log, Boolean in, WorkerTaskResult value) {
         );
     }
 
+    protected static void log(Logger log, Boolean in, Execution value) {
+        log.debug(
+            "{} {} [key='{}']\n{}",
+            in ? "<< IN " : ">> OUT",
+            value.getClass().getSimpleName(),
+            value.getId(),
+            value.toStringState()
+        );
+    }
+
     protected static void log(Logger log, Boolean in, Executor value) {
         log.debug(
             "{} {} [key='{}', from='{}', offset='{}']\n{}",
diff --git a/core/src/main/java/io/kestra/core/runners/Executor.java b/core/src/main/java/io/kestra/core/runners/Executor.java
index 11796b7d90..bb551a47f8 100644
--- a/core/src/main/java/io/kestra/core/runners/Executor.java
+++ b/core/src/main/java/io/kestra/core/runners/Executor.java
@@ -22,6 +22,7 @@ public class Executor {
     private final List<WorkerTask> workerTasks = new ArrayList<>();
     private final List<WorkerTaskResult> workerTaskResults = new ArrayList<>();
     private WorkerTaskResult joined;
+    private final List<WorkerTaskExecution> workerTaskExecutions = new ArrayList<>();
 
     public Executor(Execution execution, Long offset) {
         this.execution = execution;
@@ -75,6 +76,13 @@ public Executor withWorkerTaskResults(List<WorkerTaskResult> workerTaskResults,
         return this;
     }
 
+    public Executor withWorkerTaskExecutions(List<WorkerTaskExecution> newExecutions, String from) {
+        this.workerTaskExecutions.addAll(newExecutions);
+        this.from.add(from);
+
+        return this;
+    }
+
     public Executor serialize() {
         return new Executor(
             execution,
diff --git a/core/src/main/java/io/kestra/core/runners/WorkerTaskExecution.java b/core/src/main/java/io/kestra/core/runners/WorkerTaskExecution.java
new file mode 100644
index 0000000000..6299f7b136
--- /dev/null
+++ b/core/src/main/java/io/kestra/core/runners/WorkerTaskExecution.java
@@ -0,0 +1,25 @@
+package io.kestra.core.runners;
+
+import io.kestra.core.models.executions.Execution;
+import io.kestra.core.models.executions.TaskRun;
+import io.kestra.core.tasks.flows.Flow;
+import lombok.Builder;
+import lombok.Data;
+
+import javax.validation.constraints.NotNull;
+
+@Data
+@Builder
+public class WorkerTaskExecution {
+    @NotNull
+    private TaskRun taskRun;
+
+    @NotNull
+    private Flow task;
+
+    @NotNull
+    private Execution execution;
+
+    @NotNull
+    private RunContext runContext;
+}
diff --git a/core/src/main/java/io/kestra/core/services/FlowService.java b/core/src/main/java/io/kestra/core/services/FlowService.java
index 92a6e45d97..29ec00d896 100644
--- a/core/src/main/java/io/kestra/core/services/FlowService.java
+++ b/core/src/main/java/io/kestra/core/services/FlowService.java
@@ -82,7 +82,7 @@ public List<FlowWithFlowTrigger> flowWithFlowTrigger(Stream<Flow> flowStream) {
     }
 
     private static Stream<Flow> removeSelf(Stream<Flow> flowStream, Execution execution) {
-        // we don't allow recursive 
+        // we don't allow recursive
         return flowStream
             .filter(f -> !f.uidWithoutRevision().equals(Flow.uidWithoutRevision(execution)));
     }
@@ -159,6 +159,7 @@ public List<MultipleConditionWindow> multipleFlowTrigger(
 
                 return f.getMultipleConditionWindow().with(results);
             })
+            .filter(multipleConditionWindow -> multipleConditionWindow.getResults().size() > 0)
             .collect(Collectors.toList());
     }
 
diff --git a/core/src/main/java/io/kestra/core/tasks/flows/Flow.java b/core/src/main/java/io/kestra/core/tasks/flows/Flow.java
index 4222e63818..af21e3a9b1 100644
--- a/core/src/main/java/io/kestra/core/tasks/flows/Flow.java
+++ b/core/src/main/java/io/kestra/core/tasks/flows/Flow.java
@@ -1,28 +1,26 @@
 package io.kestra.core.tasks.flows;
 
 import com.google.common.collect.ImmutableMap;
-import io.micronaut.inject.qualifiers.Qualifiers;
-import io.swagger.v3.oas.annotations.media.Schema;
-import lombok.*;
-import lombok.experimental.SuperBuilder;
 import io.kestra.core.models.annotations.Example;
 import io.kestra.core.models.annotations.Plugin;
 import io.kestra.core.models.annotations.PluginProperty;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.executions.ExecutionTrigger;
+import io.kestra.core.models.executions.TaskRun;
 import io.kestra.core.models.flows.State;
 import io.kestra.core.models.tasks.RunnableTask;
 import io.kestra.core.models.tasks.Task;
-import io.kestra.core.queues.QueueFactoryInterface;
-import io.kestra.core.queues.QueueInterface;
 import io.kestra.core.repositories.FlowRepositoryInterface;
-import io.kestra.core.runners.RunContext;
-import io.kestra.core.runners.RunnerUtils;
+import io.kestra.core.runners.*;
+import io.swagger.v3.oas.annotations.media.Schema;
+import lombok.*;
+import lombok.experimental.SuperBuilder;
 import org.slf4j.Logger;
 
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
+import javax.annotation.Nullable;
 import javax.validation.constraints.NotNull;
 
 @SuperBuilder
@@ -92,16 +90,26 @@ public class Flow extends Task implements RunnableTask<Flow.Output> {
     @PluginProperty(dynamic = false)
     private final Boolean transmitFailed = false;
 
-    @SuppressWarnings("unchecked")
+    @Schema(
+        title = "Extract outputs from triggered executions.",
+        description = "Allow to specify key value (with value renderered), in order to extract any outputs from " +
+            "triggered execution."
+    )
+    @PluginProperty(dynamic = true)
+    private Map<String, Object> outputs;
+
     @Override
     public Flow.Output run(RunContext runContext) throws Exception {
+        throw new IllegalStateException("This task must not be run by a worker and must be run on executor side!");
+    }
+
+    @SuppressWarnings("unchecked")
+    public Execution createExecution(RunContext runContext) throws Exception {
         Logger logger = runContext.logger();
         RunnerUtils runnerUtils = runContext.getApplicationContext().getBean(RunnerUtils.class);
+
+        // @TODO: remove
         FlowRepositoryInterface flowRepository = runContext.getApplicationContext().getBean(FlowRepositoryInterface.class);
-        QueueInterface<Execution> executionQueue = (QueueInterface<Execution>) runContext.getApplicationContext().getBean(
-            QueueInterface.class,
-            Qualifiers.byName(QueueFactoryInterface.EXECUTION_NAMED)
-        );
 
         Map<String, String> inputs = new HashMap<>();
         if (this.inputs != null) {
@@ -135,9 +143,6 @@ public Flow.Output run(RunContext runContext) throws Exception {
                 .build()
             );
 
-        Output.OutputBuilder outputBuilder = Output.builder()
-            .executionId(execution.getId());
-
         logger.debug(
             "Create new execution for flow {}.{} with id {}",
             execution.getNamespace(),
@@ -145,25 +150,58 @@ public Flow.Output run(RunContext runContext) throws Exception {
             execution.getId()
         );
 
-        if (!wait) {
-            executionQueue.emit(execution);
-        } else {
-            Execution ended = runnerUtils.awaitExecution(
-                runnerUtils.isTerminatedExecution(execution, flow),
-                () -> {
-                    executionQueue.emit(execution);
-                },
-                null
-            );
+        return execution;
+    }
+
+    public WorkerTaskResult createWorkerTaskResult(
+        @Nullable RunContextFactory runContextFactory,
+        WorkerTaskExecution workerTaskExecution,
+        @Nullable io.kestra.core.models.flows.Flow flow,
+        Execution execution
+    ) {
+        TaskRun taskRun = workerTaskExecution.getTaskRun();
 
-            outputBuilder.state(ended.getState().getCurrent());
+        Output.OutputBuilder builder = Output.builder()
+            .executionId(execution.getId());
+
+        if (workerTaskExecution.getTask().getOutputs() != null && runContextFactory != null) {
+            RunContext runContext = runContextFactory.of(
+                flow,
+                workerTaskExecution.getTask(),
+                execution,
+                workerTaskExecution.getTaskRun()
+            );
 
-            if (transmitFailed && (ended.getState().isFailed() || ended.getState().getCurrent() == State.Type.KILLED)) {
-                throw new Exception("Execution '" + ended.getId() + "' failed with status '" + ended.getState().getCurrent() + "'");
+            try {
+                builder.outputs(runContext.render(workerTaskExecution.getTask().getOutputs()));
+            } catch (Exception e) {
+                runContext.logger().warn("Failed to extract ouputs with error: '" + e.getMessage() + "'", e);
+                taskRun = taskRun
+                    .withState(State.Type.FAILED)
+                    .withOutputs(builder.build().toMap());
+
+                return WorkerTaskResult.builder()
+                    .task(workerTaskExecution.getTask())
+                    .taskRun(taskRun)
+                    .build();
             }
         }
 
-        return outputBuilder
+        builder.state(execution.getState().getCurrent());
+
+        taskRun = taskRun.withOutputs(builder.build().toMap());
+
+        if (transmitFailed &&
+            (execution.getState().isFailed() || execution.getState().getCurrent() == State.Type.KILLED || execution.getState().getCurrent() == State.Type.WARNING)
+        ) {
+            taskRun = taskRun.withState(execution.getState().getCurrent());
+        } else {
+            taskRun = taskRun.withState(State.Type.SUCCESS);
+        }
+
+        return WorkerTaskResult.builder()
+            .task(workerTaskExecution.getTask())
+            .taskRun(taskRun)
             .build();
     }
 
@@ -179,6 +217,11 @@ public static class Output implements io.kestra.core.models.tasks.Output {
             title = "The state of the execution trigger.",
             description = "Only available if the execution is waited with `wait` options"
         )
-        private State.Type state;
+        private final State.Type state;
+
+        @Schema(
+            title = "The extracted outputs from triggered executions."
+        )
+        private final Map<String, Object> outputs;
     }
 }
diff --git a/runner-kafka/src/main/java/io/kestra/runner/kafka/KafkaExecutor.java b/runner-kafka/src/main/java/io/kestra/runner/kafka/KafkaExecutor.java
index b27ee260b8..dcaaf4f58e 100644
--- a/runner-kafka/src/main/java/io/kestra/runner/kafka/KafkaExecutor.java
+++ b/runner-kafka/src/main/java/io/kestra/runner/kafka/KafkaExecutor.java
@@ -159,6 +159,13 @@ public StreamsBuilder topology() {
         this.toExecutorFlowTriggerTopic(stream);
         this.handleFlowTrigger(flowWithTriggerStream);
 
+        // task Flow
+        KTable<String, WorkerTaskExecution> workerTaskExecutionKTable = this.workerTaskExecutionStream(builder);
+
+        this.toWorkerTaskExecution(stream);
+        this.workerTaskExecutionToExecution(stream);
+        this.handleWorkerTaskExecution(workerTaskExecutionKTable, stream);
+
         // purge at end
         this.purgeExecutor(stream);
 
@@ -568,36 +575,113 @@ private void toWorkerTask(KStream<String, Executor> stream) {
             );
     }
 
+    private KTable<String, WorkerTaskExecution> workerTaskExecutionStream(StreamsBuilder builder) {
+        return builder
+            .table(
+                kafkaAdminService.getTopicName(WorkerTaskExecution.class),
+                Consumed.with(Serdes.String(), JsonSerde.of(WorkerTaskExecution.class)).withName("WorkerTaskExecution.from"),
+                Materialized.<String, WorkerTaskExecution, KeyValueStore<Bytes, byte[]>>as("workertaskexecution")
+                    .withKeySerde(Serdes.String())
+                    .withValueSerde(JsonSerde.of(WorkerTaskExecution.class))
+            );
+    }
+
+    private void toWorkerTaskExecution(KStream<String, Executor> stream) {
+        stream
+            .flatMapValues(
+                (readOnlyKey, value) -> value.getWorkerTaskExecutions(),
+                Named.as("ToWorkerTaskExecution.flatMap")
+            )
+            .selectKey(
+                (key, value) -> value.getExecution().getId(),
+                Named.as("ToWorkerTaskExecution.selectKey")
+            )
+            .to(
+                kafkaAdminService.getTopicName(WorkerTaskExecution.class),
+                Produced.with(Serdes.String(), JsonSerde.of(WorkerTaskExecution.class)).withName("ToWorkerTaskExecution.toWorkerTaskExecution")
+            );
+    }
+
+    private void workerTaskExecutionToExecution(KStream<String, Executor> stream) {
+        KStream<String, Execution> executionKStream = stream
+            .flatMapValues(
+                (readOnlyKey, value) -> value.getWorkerTaskExecutions(),
+                Named.as("WorkerTaskExecutionToExecution.flatMap")
+            )
+            .mapValues(
+                (key, value) -> value.getExecution(),
+                Named.as("WorkerTaskExecutionToExecution.map")
+            )
+            .selectKey(
+                (key, value) -> value.getId(),
+                Named.as("WorkerTaskExecutionToExecution.selectKey")
+            );
+
+        executionKStream = KafkaStreamSourceService.logIfEnabled(
+            executionKStream,
+            (key, value) -> log(log, false, value),
+            "WorkerTaskExecutionToExecution"
+        );
+
+        executionKStream
+            .to(
+                kafkaAdminService.getTopicName(Execution.class),
+                Produced.with(Serdes.String(), JsonSerde.of(Execution.class)).withName("WorkerTaskExecutionToExecution.toExecution")
+            );
+    }
+
+    private void handleWorkerTaskExecution(KTable<String, WorkerTaskExecution> workerTaskExecutionKTable, KStream<String, Executor> stream) {
+        KStream<String, WorkerTaskResult> joinKStream = stream
+            .filter(
+                (key, value) -> conditionService.isTerminatedWithListeners(value.getFlow(), value.getExecution()),
+                Named.as("HandleWorkerTaskExecution.isTerminated")
+            )
+            .transformValues(
+                () -> new WorkerTaskExecutionTransformer(runContextFactory, workerTaskExecutionKTable.queryableStoreName()),
+                 Named.as("HandleWorkerTaskExecution.transform"),
+                workerTaskExecutionKTable.queryableStoreName()
+            )
+            .filter((key, value) -> value != null, Named.as("HandleWorkerTaskExecution.joinNotNullFilter"));
+
+        toWorkerTaskResultSend(joinKStream, "HandleWorkerTaskExecution");
+    }
+
     private void toWorkerTaskResult(KStream<String, Executor> stream) {
         KStream<String, WorkerTaskResult> workerTaskResultKStream = stream
             .flatMapValues(
                 (readOnlyKey, value) -> value.getWorkerTaskResults(),
                 Named.as("HandleWorkerTaskResult.flapMap")
-            )
+            );
+
+        toWorkerTaskResultSend(workerTaskResultKStream, "HandleWorkerTaskResult");
+    }
+
+    private void toWorkerTaskResultSend(KStream<String, WorkerTaskResult> stream, String name) {
+        KStream<String, WorkerTaskResult> workerTaskResultKStream = stream
             .transformValues(
                 () -> new DeduplicationTransformer<>(
-                    "WorkerTaskResult",
+                    name,
                     WORKERTASK_DEDUPLICATION_STATE_STORE_NAME,
                     (key, value) -> value.getTaskRun().getExecutionId() + "-" + value.getTaskRun().getId(),
                     (key, value) -> value.getTaskRun().getState().getCurrent().name()
                 ),
-                Named.as("HandleWorkerTaskResult.deduplication"),
+                Named.as(name + ".deduplication"),
                 WORKERTASK_DEDUPLICATION_STATE_STORE_NAME
             )
-            .filter((key, value) -> value != null, Named.as("HandleWorkerTaskResult.notNullFilter"))
+            .filter((key, value) -> value != null, Named.as(name + ".notNullFilter"))
             .selectKey(
                 (key, value) -> value.getTaskRun().getId(),
-                Named.as("HandleWorkerTaskResult.selectKey")
+                Named.as(name + ".selectKey")
             );
 
         KafkaStreamSourceService.logIfEnabled(
             workerTaskResultKStream,
             (key, value) -> log(log, false, value),
-            "HandleWorkerTaskResult"
+            name
         )
             .to(
                 kafkaAdminService.getTopicName(WorkerTaskResult.class),
-                Produced.with(Serdes.String(), JsonSerde.of(WorkerTaskResult.class)).withName("HandleWorkerTaskResult.toWorkerTaskResult")
+                Produced.with(Serdes.String(), JsonSerde.of(WorkerTaskResult.class)).withName(name + ".toWorkerTaskResult")
             );
     }
 
@@ -797,6 +881,7 @@ public void run() {
         kafkaAdminService.createIfNotExist(Executor.class);
         kafkaAdminService.createIfNotExist(KafkaStreamSourceService.TOPIC_EXECUTOR_WORKERINSTANCE);
         kafkaAdminService.createIfNotExist(ExecutionKilled.class);
+        kafkaAdminService.createIfNotExist(WorkerTaskExecution.class);
         kafkaAdminService.createIfNotExist(WorkerTaskRunning.class);
         kafkaAdminService.createIfNotExist(WorkerInstance.class);
         kafkaAdminService.createIfNotExist(Template.class);
diff --git a/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/WorkerTaskExecutionTransformer.java b/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/WorkerTaskExecutionTransformer.java
new file mode 100644
index 0000000000..c8871e973c
--- /dev/null
+++ b/runner-kafka/src/main/java/io/kestra/runner/kafka/streams/WorkerTaskExecutionTransformer.java
@@ -0,0 +1,54 @@
+package io.kestra.runner.kafka.streams;
+
+import io.kestra.core.models.flows.Flow;
+import io.kestra.core.runners.Executor;
+import io.kestra.core.runners.RunContextFactory;
+import io.kestra.core.runners.WorkerTaskExecution;
+import io.kestra.core.runners.WorkerTaskResult;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.streams.kstream.ValueTransformerWithKey;
+import org.apache.kafka.streams.processor.ProcessorContext;
+import org.apache.kafka.streams.state.KeyValueStore;
+import org.apache.kafka.streams.state.ValueAndTimestamp;
+
+@Slf4j
+public class WorkerTaskExecutionTransformer implements ValueTransformerWithKey<String, Executor, WorkerTaskResult> {
+    private final String workerTaskExecutionStoreName;
+    private final RunContextFactory runContextFactory;
+
+    private KeyValueStore<String, ValueAndTimestamp<WorkerTaskExecution>> workerTaskExecutionStore;
+    private KeyValueStore<String, ValueAndTimestamp<Flow>> flowStore;
+
+    public WorkerTaskExecutionTransformer(RunContextFactory runContextFactory, String workerTaskExecutionStoreName) {
+        this.runContextFactory = runContextFactory;
+        this.workerTaskExecutionStoreName = workerTaskExecutionStoreName;
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    public void init(final ProcessorContext context) {
+        this.flowStore = (KeyValueStore<String, ValueAndTimestamp<Flow>>) context.getStateStore("flow");
+        this.workerTaskExecutionStore = (KeyValueStore<String, ValueAndTimestamp<WorkerTaskExecution>>) context.getStateStore(this.workerTaskExecutionStoreName);
+    }
+
+    @Override
+    public WorkerTaskResult transform(final String key, final Executor value) {
+        ValueAndTimestamp<WorkerTaskExecution> workerTaskExecutionStoreValue = workerTaskExecutionStore.get(key);
+        if (workerTaskExecutionStoreValue == null) {
+            return null;
+        }
+
+        WorkerTaskExecution workerTaskExecution = workerTaskExecutionStoreValue.value();
+
+        ValueAndTimestamp<Flow> flowValueAndTimestamp = this.flowStore.get(Flow.uid(value.getExecution()));
+        Flow flow = flowValueAndTimestamp.value();
+
+        return workerTaskExecution
+            .getTask()
+            .createWorkerTaskResult(runContextFactory, workerTaskExecution, flow, value.getExecution());
+    }
+
+    @Override
+    public void close() {
+    }
+}
diff --git a/runner-memory/src/main/java/io/kestra/runner/memory/MemoryExecutor.java b/runner-memory/src/main/java/io/kestra/runner/memory/MemoryExecutor.java
index 06d872db08..de40aba513 100644
--- a/runner-memory/src/main/java/io/kestra/runner/memory/MemoryExecutor.java
+++ b/runner-memory/src/main/java/io/kestra/runner/memory/MemoryExecutor.java
@@ -38,7 +38,9 @@ public class MemoryExecutor extends AbstractExecutor {
 
     private static final MemoryMultipleConditionStorage multipleConditionStorage = new MemoryMultipleConditionStorage();
 
-    private static final ConcurrentHashMap<String, ExecutionState> executions = new ConcurrentHashMap<>();
+    private static final ConcurrentHashMap<String, ExecutionState> EXECUTIONS = new ConcurrentHashMap<>();
+    private static final ConcurrentHashMap<String, WorkerTaskExecution> WORKERTASKEXECUTIONS_WATCHER = new ConcurrentHashMap<>();
+
     private List<Flow> allFlows;
 
     @Inject
@@ -130,11 +132,21 @@ private void handleExecution(ExecutionState state) {
                     .forEach(workerTaskResultQueue::emit);
             }
 
+            if (executor.getWorkerTaskExecutions().size() > 0) {
+                executor.getWorkerTaskExecutions()
+                    .forEach(workerTaskExecution -> {
+                        WORKERTASKEXECUTIONS_WATCHER.put(workerTaskExecution.getExecution().getId(), workerTaskExecution);
+
+                        executionQueue.emit(workerTaskExecution.getExecution());
+                    });
+            }
+
             // Listeners need the last emit
             if (conditionService.isTerminatedWithListeners(flow, execution)) {
                 this.executionQueue.emit(execution);
             }
 
+            // multiple condition
             if (conditionService.isTerminatedWithListeners(flow, execution)) {
                 // multiple conditions storage
                 multipleConditionStorage.save(
@@ -152,6 +164,20 @@ private void handleExecution(ExecutionState state) {
                     .multipleFlowToDelete(allFlows.stream(), multipleConditionStorage)
                     .forEach(multipleConditionStorage::delete);
             }
+
+            // worker task execution
+            if (conditionService.isTerminatedWithListeners(flow, execution) && WORKERTASKEXECUTIONS_WATCHER.containsKey(execution.getId())) {
+                WorkerTaskExecution workerTaskExecution = WORKERTASKEXECUTIONS_WATCHER.get(execution.getId());
+                Flow workerTaskFlow = this.flowRepository.findByExecution(execution);
+
+                WorkerTaskResult workerTaskResult = workerTaskExecution
+                    .getTask()
+                    .createWorkerTaskResult(runContextFactory, workerTaskExecution, workerTaskFlow, execution);
+
+                this.workerTaskResultQueue.emit(workerTaskResult);
+
+                WORKERTASKEXECUTIONS_WATCHER.remove(execution.getId());
+            }
         }
     }
 
@@ -168,7 +194,7 @@ private void handleFailedExecutionFromExecutor(Executor executor, Exception e) {
 
     private ExecutionState saveExecution(Execution execution) {
         ExecutionState queued;
-        queued = executions.compute(execution.getId(), (s, executionState) -> {
+        queued = EXECUTIONS.compute(execution.getId(), (s, executionState) -> {
             if (executionState == null) {
                 return new ExecutionState(execution);
             } else {
@@ -192,7 +218,7 @@ private void toExecution(Executor executor) {
 
         // delete if ended
         if (conditionService.isTerminatedWithListeners(executor.getFlow(), executor.getExecution())) {
-            executions.remove(executor.getExecution().getId());
+            EXECUTIONS.remove(executor.getExecution().getId());
         }
     }
 
@@ -211,7 +237,7 @@ private void workerTaskResultQueue(WorkerTaskResult message) {
                 .record(message.getTaskRun().getState().getDuration());
 
             // save WorkerTaskResult on current QueuedExecution
-            executions.compute(message.getTaskRun().getExecutionId(), (s, executionState) -> {
+            EXECUTIONS.compute(message.getTaskRun().getExecutionId(), (s, executionState) -> {
                 if (executionState == null) {
                     throw new IllegalStateException("Execution state don't exist for " + s + ", receive " + message);
                 }
@@ -223,16 +249,15 @@ private void workerTaskResultQueue(WorkerTaskResult message) {
                 }
             });
 
+            Flow flow = this.flowRepository.findByExecution(EXECUTIONS.get(message.getTaskRun().getExecutionId()).execution);
+            flow = taskDefaultService.injectDefaults(flow, EXECUTIONS.get(message.getTaskRun().getExecutionId()).execution);
 
-            Flow flow = this.flowRepository.findByExecution(executions.get(message.getTaskRun().getExecutionId()).execution);
-            flow = taskDefaultService.injectDefaults(flow, executions.get(message.getTaskRun().getExecutionId()).execution);
-
-            this.toExecution(new Executor(executions.get(message.getTaskRun().getExecutionId()).execution, null).withFlow(flow));
+            this.toExecution(new Executor(EXECUTIONS.get(message.getTaskRun().getExecutionId()).execution, null).withFlow(flow));
         }
     }
 
     private boolean deduplicateWorkerTask(Execution execution, TaskRun taskRun) {
-        ExecutionState executionState = executions.get(execution.getId());
+        ExecutionState executionState = EXECUTIONS.get(execution.getId());
 
         String deduplicationKey = taskRun.getExecutionId() + "-" + taskRun.getId();
         State.Type current = executionState.workerTaskDeduplication.get(deduplicationKey);
@@ -247,7 +272,7 @@ private boolean deduplicateWorkerTask(Execution execution, TaskRun taskRun) {
     }
 
     private boolean deduplicateNexts(Execution execution, List<TaskRun> taskRuns) {
-        ExecutionState executionState = executions.get(execution.getId());
+        ExecutionState executionState = EXECUTIONS.get(execution.getId());
 
         return taskRuns
             .stream()


Test patch:
diff --git a/core/src/test/java/io/kestra/core/Helpers.java b/core/src/test/java/io/kestra/core/Helpers.java
index f1e3197f28..5fff908779 100644
--- a/core/src/test/java/io/kestra/core/Helpers.java
+++ b/core/src/test/java/io/kestra/core/Helpers.java
@@ -17,7 +17,7 @@
 import java.util.function.BiConsumer;
 
 public class Helpers {
-    public static long FLOWS_COUNT = 45;
+    public static long FLOWS_COUNT = 46;
 
     public static ApplicationContext applicationContext() throws URISyntaxException {
         return applicationContext(Paths.get(Objects.requireNonNull(Helpers.class.getClassLoader().getResource("plugins")).toURI()));
diff --git a/core/src/test/java/io/kestra/core/tasks/flows/FlowCaseTest.java b/core/src/test/java/io/kestra/core/tasks/flows/FlowCaseTest.java
new file mode 100644
index 0000000000..e0e2220169
--- /dev/null
+++ b/core/src/test/java/io/kestra/core/tasks/flows/FlowCaseTest.java
@@ -0,0 +1,86 @@
+package io.kestra.core.tasks.flows;
+
+import com.google.common.collect.ImmutableMap;
+import io.kestra.core.models.executions.Execution;
+import io.kestra.core.models.flows.State;
+import io.kestra.core.queues.QueueFactoryInterface;
+import io.kestra.core.queues.QueueInterface;
+import io.kestra.core.runners.RunnerUtils;
+
+import java.time.Duration;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import javax.inject.Inject;
+import javax.inject.Named;
+import javax.inject.Singleton;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.*;
+
+@Singleton
+public class FlowCaseTest {
+    @Inject
+    @Named(QueueFactoryInterface.EXECUTION_NAMED)
+    QueueInterface<Execution> executionQueue;
+
+    @Inject
+    protected RunnerUtils runnerUtils;
+
+    public void waitSuccess() throws Exception {
+        this.run("OK", State.Type.SUCCESS, State.Type.SUCCESS, 2, "default >");
+    }
+
+    public void waitFailed() throws Exception {
+        this.run("THIRD", State.Type.FAILED, State.Type.FAILED, 4, "Error Trigger ! error-t1");
+    }
+
+    public void invalidOutputs() throws Exception {
+        this.run("FIRST", State.Type.FAILED, State.Type.SUCCESS, 2, null);
+    }
+
+    @SuppressWarnings({"ResultOfMethodCallIgnored", "unchecked"})
+    void run(String input, State.Type fromState, State.Type triggerState,  int count, String outputs) throws Exception {
+        CountDownLatch countDownLatch = new CountDownLatch(1);
+        AtomicReference<Execution> triggered = new AtomicReference<>();
+
+        executionQueue.receive(execution -> {
+            if (execution.getFlowId().equals("switch") && execution.getState().getCurrent().isTerninated()) {
+                countDownLatch.countDown();
+                triggered.set(execution);
+            }
+        });
+
+        Execution execution = runnerUtils.runOne(
+            "io.kestra.tests",
+            "task-flow",
+            null,
+            (f, e) -> ImmutableMap.of("string", input),
+            Duration.ofMinutes(1)
+        );
+
+        countDownLatch.await(1, TimeUnit.MINUTES);
+
+        assertThat(execution.getTaskRunList(), hasSize(1));
+        assertThat(execution.getState().getCurrent(), is(fromState));
+
+        if (outputs != null) {
+            assertThat(((Map<String, String>) execution.getTaskRunList().get(0).getOutputs().get("outputs")).get("extracted"), containsString(outputs));
+        }
+
+        assertThat(execution.getTaskRunList().get(0).getOutputs().get("executionId"), is(triggered.get().getId()));
+
+        if (outputs != null) {
+            assertThat(execution.getTaskRunList().get(0).getOutputs().get("state"), is(triggered.get().getState().getCurrent().name()));
+        }
+
+        assertThat(triggered.get().getTrigger().getType(), is(Flow.class.getName()));
+        assertThat(triggered.get().getTrigger().getVariables().get("executionId"), is(execution.getId()));
+        assertThat(triggered.get().getTrigger().getVariables().get("flowId"), is(execution.getFlowId()));
+        assertThat(triggered.get().getTrigger().getVariables().get("namespace"), is(execution.getNamespace()));
+
+        assertThat(triggered.get().getTaskRunList(), hasSize(count));
+        assertThat(triggered.get().getState().getCurrent(), is(triggerState));
+    }
+}
diff --git a/core/src/test/java/io/kestra/core/tasks/flows/FlowTest.java b/core/src/test/java/io/kestra/core/tasks/flows/FlowTest.java
index 2f86806140..1e40362a9c 100644
--- a/core/src/test/java/io/kestra/core/tasks/flows/FlowTest.java
+++ b/core/src/test/java/io/kestra/core/tasks/flows/FlowTest.java
@@ -1,66 +1,26 @@
 package io.kestra.core.tasks.flows;
 
-import com.google.common.collect.ImmutableMap;
+import io.kestra.core.runners.AbstractMemoryRunnerTest;
 import org.junit.jupiter.api.Test;
-import io.kestra.core.models.executions.Execution;
-import io.kestra.core.models.flows.State;
-import io.kestra.core.queues.QueueFactoryInterface;
-import io.kestra.core.queues.QueueInterface;
-import io.kestra.core.repositories.ExecutionRepositoryInterface;
-import io.kestra.core.repositories.FlowRepositoryInterface;
-import io.kestra.core.runners.*;
-import io.kestra.core.utils.TestsUtils;
 
-import java.util.Map;
-import java.util.Optional;
 import javax.inject.Inject;
-import javax.inject.Named;
-
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.is;
-
-class FlowTest extends AbstractMemoryRunnerTest {
-    @Inject
-    RunContextFactory runContextFactory;
-
-    @Inject
-    FlowRepositoryInterface flowRepositoryInterface;
-
-    @Inject
-    @Named(QueueFactoryInterface.EXECUTION_NAMED)
-    QueueInterface<Execution> executionQueue;
 
+public class FlowTest extends AbstractMemoryRunnerTest {
     @Inject
-    ExecutionRepositoryInterface executionRepository;
+    FlowCaseTest flowCaseTest;
 
-    @SuppressWarnings("unchecked")
     @Test
-    void run() throws Exception {
-        Flow flow = Flow.builder()
-            .id("unit-test")
-            .type(Flow.class.getName())
-            .namespace("{{inputs.namespace}}")
-            .flowId("{{inputs.flow}}")
-            .inputs(InputsTest.inputs)
-            .wait(true)
-            .build();
-
-        RunContext runContext = TestsUtils.mockRunContext(runContextFactory, flow, ImmutableMap.of(
-            "namespace", "io.kestra.tests",
-            "flow", "inputs"
-        ));
-
-        Flow.Output run = flow.run(runContext);
-
-        assertThat(run.getExecutionId() == null, is(false));
-        assertThat(run.getState(), is(State.Type.SUCCESS));
+    public void waitSuccess() throws Exception {
+        flowCaseTest.waitSuccess();
+    }
 
-        Optional<Execution> execution = executionRepository.findById(run.getExecutionId());
+    @Test
+    public void waitFailed() throws Exception {
+        flowCaseTest.waitFailed();
+    }
 
-        assertThat(execution.isPresent(), is(true));
-        assertThat(execution.get().getTrigger().getType(), is(Flow.class.getName()));
-        assertThat(execution.get().getTrigger().getVariables().get("executionId"), is(((Map<String, String>) runContext.getVariables().get("execution")).get("id")));
-        assertThat(execution.get().getTrigger().getVariables().get("flowId"), is(((Map<String, String>) runContext.getVariables().get("flow")).get("id")));
-        assertThat(execution.get().getTrigger().getVariables().get("namespace"), is(((Map<String, String>) runContext.getVariables().get("flow")).get("namespace")));
+    @Test
+    public void invalidOutputs() throws Exception {
+        flowCaseTest.invalidOutputs();
     }
 }
diff --git a/core/src/test/resources/flows/valids/task-flow.yaml b/core/src/test/resources/flows/valids/task-flow.yaml
new file mode 100644
index 0000000000..6622ebe3b8
--- /dev/null
+++ b/core/src/test/resources/flows/valids/task-flow.yaml
@@ -0,0 +1,18 @@
+id: task-flow
+namespace: io.kestra.tests
+
+inputs:
+  - name: string
+    type: STRING
+
+tasks:
+  - id: launch
+    type: io.kestra.core.tasks.flows.Flow
+    namespace: io.kestra.tests
+    flowId: switch
+    inputs:
+      string: "{{ inputs.string }}"
+    wait: true
+    transmitFailed: true
+    outputs:
+      extracted: "{{ firstDefined outputs.default.value outputs.error-t1.value }}"
\ No newline at end of file
diff --git a/runner-kafka/src/test/java/io/kestra/runner/kafka/KafkaRunnerTest.java b/runner-kafka/src/test/java/io/kestra/runner/kafka/KafkaRunnerTest.java
index 6ba02124d6..16224341df 100644
--- a/runner-kafka/src/test/java/io/kestra/runner/kafka/KafkaRunnerTest.java
+++ b/runner-kafka/src/test/java/io/kestra/runner/kafka/KafkaRunnerTest.java
@@ -10,10 +10,10 @@
 import io.kestra.core.repositories.TemplateRepositoryInterface;
 import io.kestra.core.runners.*;
 import io.kestra.core.tasks.flows.EachSequentialTest;
+import io.kestra.core.tasks.flows.FlowCaseTest;
 import io.kestra.core.tasks.flows.TemplateTest;
 import io.kestra.core.utils.TestsUtils;
 import org.apache.kafka.common.errors.RecordTooLargeException;
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 
 import java.io.IOException;
@@ -46,6 +46,9 @@ class KafkaRunnerTest extends AbstractKafkaRunnerTest {
     @Inject
     private TemplateRepositoryInterface templateRepository;
 
+    @Inject
+    private FlowCaseTest flowCaseTest;
+
     @Inject
     @Named(QueueFactoryInterface.WORKERTASKLOG_NAMED)
     private QueueInterface<LogEntry> workerTaskLogQueue;
@@ -251,4 +254,19 @@ void invalidTaskDefaults() throws TimeoutException, IOException, URISyntaxExcept
         repositoryLoader.load(Objects.requireNonNull(ListenersTest.class.getClassLoader().getResource("flows/tests/invalid-task-defaults.yaml")));
         taskDefaultsCaseTest.invalidTaskDefaults();
     }
+
+    @Test
+    void flowWaitSuccess() throws Exception {
+        flowCaseTest.waitSuccess();
+    }
+
+    @Test
+    void flowWaitFailed() throws Exception {
+        flowCaseTest.waitFailed();
+    }
+
+    @Test
+    public void invalidOutputs() throws Exception {
+        flowCaseTest.invalidOutputs();
+    }
 }
diff --git a/runner-kafka/src/test/resources/application.yml b/runner-kafka/src/test/resources/application.yml
index a150251734..9528712396 100644
--- a/runner-kafka/src/test/resources/application.yml
+++ b/runner-kafka/src/test/resources/application.yml
@@ -93,6 +93,12 @@ kestra:
         properties:
           cleanup.policy: "delete,compact"
 
+      executorworkertaskexecution:
+        cls: io.kestra.core.runners.WorkerTaskExecution
+        name: "${kestra.kafka.defaults.topic-prefix}executor_workertaskexecution"
+        properties:
+          cleanup.policy: "delete,compact"
+
       workertask:
         name: "${kestra.kafka.defaults.topic-prefix}workertask"
         cls: io.kestra.core.runners.WorkerTask