Repository: kestra-io/kestra
Issue: taskDefault are not applied on Flowable task
Labels: bug
backend


Issue comments:

Pull request review comments:

Patch that resolved the issue:
diff --git a/cli/src/main/java/io/kestra/cli/commands/sys/RestoreQueueCommand.java b/cli/src/main/java/io/kestra/cli/commands/sys/RestoreQueueCommand.java
index 9b88631973..cbea85bac5 100644
--- a/cli/src/main/java/io/kestra/cli/commands/sys/RestoreQueueCommand.java
+++ b/cli/src/main/java/io/kestra/cli/commands/sys/RestoreQueueCommand.java
@@ -1,31 +1,17 @@
 package io.kestra.cli.commands.sys;
 
+import io.kestra.cli.AbstractCommand;
+import io.kestra.cli.services.RestoreQueueService;
 import io.micronaut.context.ApplicationContext;
-import io.micronaut.inject.qualifiers.Qualifiers;
-import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
-import io.kestra.cli.AbstractCommand;
-import io.kestra.core.models.flows.Flow;
-import io.kestra.core.models.templates.Template;
-import io.kestra.core.models.triggers.Trigger;
-import io.kestra.core.queues.QueueFactoryInterface;
-import io.kestra.core.queues.QueueInterface;
-import io.kestra.core.repositories.FlowRepositoryInterface;
-import io.kestra.core.repositories.TemplateRepositoryInterface;
-import io.kestra.core.repositories.TriggerRepositoryInterface;
-import io.kestra.runner.kafka.services.KafkaAdminService;
 import picocli.CommandLine;
 
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Optional;
-import java.util.stream.Collectors;
 import javax.inject.Inject;
 
 @CommandLine.Command(
     name = "restore-queue",
     description = {"send all data from a repository to kafka.",
-        "Mostly usefull to send all flows, templates & triggers from repository to kafka in case of restore."
+        "Mostly useful to send all flows, templates & triggers from repository to kafka in case of restore."
     }
 )
 @Slf4j
@@ -36,13 +22,13 @@ public class RestoreQueueCommand extends AbstractCommand {
     @CommandLine.Option(names = {"--no-recreate"}, description = "Don't drop the topic and recreate it")
     private boolean noRecreate = false;
 
-    @CommandLine.Option(names = {"--no-flows"}, description = "Don't send flow")
+    @CommandLine.Option(names = {"--no-flows"}, description = "Don't send flows")
     private boolean noFlows = false;
 
-    @CommandLine.Option(names = {"--no-templates"}, description = "Don't send flow")
+    @CommandLine.Option(names = {"--no-templates"}, description = "Don't send templates")
     private boolean noTemplates = false;
 
-    @CommandLine.Option(names = {"--no-triggers"}, description = "Don't send flow")
+    @CommandLine.Option(names = {"--no-triggers"}, description = "Don't send triggers")
     private boolean noTriggers = false;
 
     public RestoreQueueCommand() {
@@ -53,57 +39,26 @@ public RestoreQueueCommand() {
     public Integer call() throws Exception {
         super.call();
 
+        RestoreQueueService restoreQueueService = applicationContext.getBean(RestoreQueueService.class);
+
         // flows
         if (!noFlows) {
-            FlowRepositoryInterface flowRepository = applicationContext.getBean(FlowRepositoryInterface.class);
-            List<Flow> flows = flowRepository
-                .findAll()
-                .stream()
-                .flatMap(flow -> flowRepository.findRevisions(flow.getNamespace(), flow.getId()).stream())
-                .collect(Collectors.toList());
-            this.send(flows, QueueFactoryInterface.FLOW_NAMED, Flow.class);
+            int size = restoreQueueService.flows(noRecreate);
+            stdOut("Successfully send {0} flows", size);
         }
 
         // templates
         if (!this.noTemplates) {
-            TemplateRepositoryInterface templateRepository = applicationContext.getBean(TemplateRepositoryInterface.class);
-            List<Template> templates = new ArrayList<>(templateRepository.findAll());
-            this.send(templates, QueueFactoryInterface.TEMPLATE_NAMED, Template.class);
+            int size = restoreQueueService.templates(noRecreate);
+            stdOut("Successfully send {0} templates", size);
         }
 
         // trigger
         if (!this.noTriggers) {
-            TriggerRepositoryInterface triggerRepository = applicationContext.getBean(TriggerRepositoryInterface.class);
-            List<Trigger> triggers = new ArrayList<>(triggerRepository.findAll());
-            this.send(triggers, QueueFactoryInterface.TRIGGER_NAMED, Trigger.class);
+            int size = restoreQueueService.triggers(noRecreate);
+            stdOut("Successfully send {0} triggers", size);
         }
 
         return 0;
     }
-
-    @SneakyThrows
-    @SuppressWarnings("unchecked")
-    private <T> void send(List<T> list, String queueName, Class<?> cls) {
-        Optional<String> queueType = applicationContext.getProperty("kestra.queue.type", String.class);
-
-        if (queueType.isPresent() && queueType.get().equals("kafka")) {
-            KafkaAdminService kafkaAdminService = applicationContext.getBean(KafkaAdminService.class);
-            if (!noRecreate) {
-                kafkaAdminService.delete(cls);
-            }
-
-            // need some wait to be sure the topic are deleted before recreated with right configuration
-            Thread.sleep(2000);
-            kafkaAdminService.createIfNotExist(cls);
-        }
-
-        QueueInterface<T> queue = (QueueInterface<T>) applicationContext.getBean(
-            QueueInterface.class,
-            Qualifiers.byName(queueName)
-        );
-
-        list.forEach(queue::emit);
-
-        stdOut("Successfully send {0} flows to {1}", list.size(), queueName);
-    }
 }
diff --git a/cli/src/main/java/io/kestra/cli/services/RestoreQueueService.java b/cli/src/main/java/io/kestra/cli/services/RestoreQueueService.java
new file mode 100644
index 0000000000..4abb53031f
--- /dev/null
+++ b/cli/src/main/java/io/kestra/cli/services/RestoreQueueService.java
@@ -0,0 +1,79 @@
+package io.kestra.cli.services;
+
+import io.kestra.core.models.flows.Flow;
+import io.kestra.core.models.templates.Template;
+import io.kestra.core.models.triggers.Trigger;
+import io.kestra.core.queues.QueueFactoryInterface;
+import io.kestra.core.queues.QueueInterface;
+import io.kestra.core.repositories.FlowRepositoryInterface;
+import io.kestra.core.repositories.TemplateRepositoryInterface;
+import io.kestra.core.repositories.TriggerRepositoryInterface;
+import io.kestra.runner.kafka.services.KafkaAdminService;
+import io.micronaut.context.ApplicationContext;
+import io.micronaut.inject.qualifiers.Qualifiers;
+import lombok.SneakyThrows;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import javax.inject.Inject;
+import javax.inject.Singleton;
+
+@Singleton
+public class RestoreQueueService {
+    @Inject
+    ApplicationContext applicationContext;
+
+    public int flows(boolean noRecreate) {
+        FlowRepositoryInterface flowRepository = applicationContext.getBean(FlowRepositoryInterface.class);
+
+        List<Flow> flows = flowRepository
+            .findAll()
+            .stream()
+            .flatMap(flow -> flowRepository.findRevisions(flow.getNamespace(), flow.getId()).stream())
+            .collect(Collectors.toList());
+
+        return this.send(flows, QueueFactoryInterface.FLOW_NAMED, Flow.class, noRecreate);
+    }
+
+    public int templates(boolean noRecreate) {
+        TemplateRepositoryInterface templateRepository = applicationContext.getBean(TemplateRepositoryInterface.class);
+        List<Template> templates = new ArrayList<>(templateRepository.findAll());
+
+        return this.send(templates, QueueFactoryInterface.TEMPLATE_NAMED, Template.class, noRecreate);
+    }
+
+    public int triggers(boolean noRecreate) {
+        TriggerRepositoryInterface triggerRepository = applicationContext.getBean(TriggerRepositoryInterface.class);
+        List<Trigger> triggers = new ArrayList<>(triggerRepository.findAll());
+
+        return this.send(triggers, QueueFactoryInterface.TRIGGER_NAMED, Trigger.class, noRecreate);
+    }
+
+    @SneakyThrows
+    @SuppressWarnings("unchecked")
+    public <T> int send(List<T> list, String queueName, Class<?> cls, boolean noRecreate) {
+        Optional<String> queueType = applicationContext.getProperty("kestra.queue.type", String.class);
+
+        if (queueType.isPresent() && queueType.get().equals("kafka")) {
+            KafkaAdminService kafkaAdminService = applicationContext.getBean(KafkaAdminService.class);
+            if (!noRecreate) {
+                kafkaAdminService.delete(cls);
+            }
+
+            // need some wait to be sure the topic are deleted before recreated with right configuration
+            Thread.sleep(2000);
+            kafkaAdminService.createIfNotExist(cls);
+        }
+
+        QueueInterface<T> queue = (QueueInterface<T>) applicationContext.getBean(
+            QueueInterface.class,
+            Qualifiers.byName(queueName)
+        );
+
+        list.forEach(queue::emit);
+
+        return list.size();
+    }
+}
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 c921dc2bd7..eb4f4fdbca 100644
--- a/core/src/main/java/io/kestra/core/runners/AbstractExecutor.java
+++ b/core/src/main/java/io/kestra/core/runners/AbstractExecutor.java
@@ -1,7 +1,6 @@
 package io.kestra.core.runners;
 
 import com.google.common.collect.ImmutableMap;
-import lombok.extern.slf4j.Slf4j;
 import io.kestra.core.exceptions.InternalException;
 import io.kestra.core.metrics.MetricRegistry;
 import io.kestra.core.models.executions.Execution;
@@ -13,7 +12,7 @@
 import io.kestra.core.models.tasks.ResolvedTask;
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.services.ConditionService;
-import io.kestra.core.services.TaskDefaultService;
+import lombok.extern.slf4j.Slf4j;
 import org.slf4j.Logger;
 
 import java.io.Closeable;
@@ -29,18 +28,15 @@ public abstract class AbstractExecutor implements Runnable, Closeable {
     protected RunContextFactory runContextFactory;
     protected MetricRegistry metricRegistry;
     protected ConditionService conditionService;
-    protected TaskDefaultService taskDefaultService;
 
     public AbstractExecutor(
         RunContextFactory runContextFactory,
         MetricRegistry metricRegistry,
-        ConditionService conditionService,
-        TaskDefaultService taskDefaultService
+        ConditionService conditionService
     ) {
         this.runContextFactory = runContextFactory;
         this.metricRegistry = metricRegistry;
         this.conditionService = conditionService;
-        this.taskDefaultService = taskDefaultService;
     }
 
     protected Execution onNexts(Flow flow, Execution execution, List<TaskRun> nexts) {
@@ -471,12 +467,10 @@ private List<WorkerTask> handleWorkerTask(Execution execution, Flow flow) throws
             .map(throwFunction(taskRun -> {
                 Task task = flow.findTaskByTaskId(taskRun.getTaskId());
 
-                Task taskWithDefault = taskDefaultService.injectDefaults(task, flow);
-
                 return WorkerTask.builder()
-                    .runContext(runContextFactory.of(flow, taskWithDefault, execution, taskRun))
+                    .runContext(runContextFactory.of(flow, task, execution, taskRun))
                     .taskRun(taskRun)
-                    .task(taskWithDefault)
+                    .task(task)
                     .build();
                 }))
             .collect(Collectors.toList());
diff --git a/core/src/main/java/io/kestra/core/services/TaskDefaultService.java b/core/src/main/java/io/kestra/core/services/TaskDefaultService.java
index 13aa3ad0ce..1ed1e0b70b 100644
--- a/core/src/main/java/io/kestra/core/services/TaskDefaultService.java
+++ b/core/src/main/java/io/kestra/core/services/TaskDefaultService.java
@@ -2,15 +2,12 @@
 
 import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.flows.TaskDefault;
-import io.kestra.core.models.tasks.Task;
 import io.kestra.core.serializers.JacksonMapper;
 import io.kestra.core.utils.MapUtils;
+import io.micronaut.core.annotation.Nullable;
 
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.stream.Collectors;
-import io.micronaut.core.annotation.Nullable;
 import javax.inject.Inject;
 import javax.inject.Singleton;
 
@@ -20,35 +17,97 @@ public class TaskDefaultService {
     @Inject
     protected TaskGlobalDefaultConfiguration globalDefault;
 
-    @SuppressWarnings("unchecked")
-    public <T extends Task> T injectDefaults(T task, Flow flow) {
-        Map<String, Object> taskAsMap = JacksonMapper.toMap(task);
+    /**
+     * @param flow the flow to extract default
+     * @return list of {@code TaskDefault} order by most important first
+     */
+    protected List<TaskDefault> mergeAllDefaults(Flow flow) {
+        List<TaskDefault> list = new ArrayList<>();
+
+        if (globalDefault != null && globalDefault.getDefaults() != null) {
+            list.addAll(globalDefault.getDefaults());
+        }
 
-        taskAsMap = defaults(task, taskAsMap, flow.getTaskDefaults());
-        if (globalDefault != null) {
-            taskAsMap = defaults(task, taskAsMap, globalDefault.getDefaults());
+        if (flow.getTaskDefaults() != null) {
+            list.addAll(flow.getTaskDefaults());
         }
 
-        return (T) JacksonMapper.toMap(taskAsMap, task.getClass());
+        return list;
     }
 
-    protected <T extends Task> List<TaskDefault> find(T task, List<TaskDefault> defaults) {
-        return (defaults == null ? new ArrayList<TaskDefault>() : defaults)
+    private static Map<String, Map<String, Object>> taskDefaultsToMap(List<TaskDefault> taskDefaults) {
+        return taskDefaults
             .stream()
-            .filter(t -> t.getType().equals(task.getType()))
-            .collect(Collectors.toList());
+            .map(taskDefault -> new AbstractMap.SimpleEntry<>(
+                taskDefault.getType(),
+                taskDefault.getValues()
+            ))
+            .collect(Collectors.toMap(
+                AbstractMap.SimpleEntry::getKey,
+                AbstractMap.SimpleEntry::getValue,
+                MapUtils::merge
+            ));
     }
 
-    protected <T extends Task> Map<String, Object> defaults(T task, Map<String, Object> taskAsMap, List<TaskDefault> defaults) {
-        for (TaskDefault current : find(task, defaults)) {
-            taskAsMap = defaults(current.getValues(), taskAsMap);
+    @SuppressWarnings("unchecked")
+    public Flow injectDefaults(Flow flow) {
+        Map<String, Object> flowAsMap = JacksonMapper.toMap(flow);
+
+        Map<String, Map<String, Object>> defaults = taskDefaultsToMap(mergeAllDefaults(flow));
+
+        Object taskDefaults = flowAsMap.get("taskDefaults");
+        if (taskDefaults != null) {
+            flowAsMap.remove("taskDefaults");
+        }
+
+        Map<String, Object> flowAsMapWithDefault = (Map<String, Object>) recursiveDefaults(flowAsMap, defaults);
+
+        if (taskDefaults != null) {
+            flowAsMapWithDefault.put("taskDefaults", taskDefaults);
         }
 
-        return taskAsMap;
+        return JacksonMapper.toMap(flowAsMapWithDefault, Flow.class);
     }
 
-    protected Map<String, Object> defaults(Map<String, Object> task, Map<String, Object> defaults) {
-        return MapUtils.merge(task, defaults);
+    private static Object recursiveDefaults(Object object, Map<String, Map<String, Object>> defaults) {
+        if (object instanceof Map) {
+            Map<?, ?> value = (Map<?, ?>) object;
+            if (value.containsKey("type")) {
+                value = defaults(value, defaults);
+            }
+
+            return value
+                .entrySet()
+                .stream()
+                .map(e -> new AbstractMap.SimpleEntry<>(
+                    e.getKey(),
+                    recursiveDefaults(e.getValue(), defaults)
+                ))
+                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+        } else if (object instanceof Collection) {
+            Collection<?> value = (Collection<?>) object;
+            return value
+                .stream()
+                .map(r -> recursiveDefaults(r, defaults))
+                .collect(Collectors.toList());
+        } else {
+            return object;
+        }
     }
 
+    @SuppressWarnings("unchecked")
+    protected static Map<?, ?> defaults(Map<?, ?> task, Map<String, Map<String, Object>> defaults) {
+        Object type = task.get("type");
+        if (!(type instanceof String)) {
+            return task;
+        }
+
+        String taskType = (String) type;
+
+        if (!defaults.containsKey(taskType)) {
+            return task;
+        }
+
+        return MapUtils.merge(defaults.get(taskType), (Map<String, Object>) task);
+    }
 }
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 1182de456d..75d81878dd 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
@@ -16,7 +16,6 @@
 import io.kestra.core.runners.*;
 import io.kestra.core.services.ConditionService;
 import io.kestra.core.services.FlowService;
-import io.kestra.core.services.TaskDefaultService;
 import io.kestra.core.utils.Either;
 import io.kestra.runner.kafka.serializers.JsonSerde;
 import io.kestra.runner.kafka.services.KafkaAdminService;
@@ -60,15 +59,15 @@ public class KafkaExecutor extends AbstractExecutor implements Closeable {
     public static final String WORKERINSTANCE_STATE_STORE_NAME = "worker_instance";
     public static final String TOPIC_EXECUTOR_WORKERINSTANCE = "executorworkerinstance";
 
-    ApplicationContext applicationContext;
-    KafkaStreamService kafkaStreamService;
-    KafkaAdminService kafkaAdminService;
-    QueueInterface<LogEntry> logQueue;
-    FlowService flowService;
-    KafkaStreamSourceService kafkaStreamSourceService;
-    QueueService queueService;
+    protected ApplicationContext applicationContext;
+    protected KafkaStreamService kafkaStreamService;
+    protected KafkaAdminService kafkaAdminService;
+    protected QueueInterface<LogEntry> logQueue;
+    protected FlowService flowService;
+    protected KafkaStreamSourceService kafkaStreamSourceService;
+    protected QueueService queueService;
 
-    KafkaStreamService.Stream resultStream;
+    protected KafkaStreamService.Stream resultStream;
     boolean ready = false;
 
     @Inject
@@ -82,10 +81,9 @@ public KafkaExecutor(
         FlowService flowService,
         ConditionService conditionService,
         KafkaStreamSourceService kafkaStreamSourceService,
-        TaskDefaultService taskDefaultService,
         QueueService queueService
     ) {
-        super(runContextFactory, metricRegistry, conditionService, taskDefaultService);
+        super(runContextFactory, metricRegistry, conditionService);
 
         this.applicationContext = applicationContext;
         this.kafkaStreamService = kafkaStreamService;
@@ -96,7 +94,7 @@ public KafkaExecutor(
         this.queueService = queueService;
     }
 
-    public Topology topology() {
+    public StreamsBuilder topology() {
         StreamsBuilder builder = new KafkaStreamsBuilder();
 
         // copy execution to be done on executor queue
@@ -189,14 +187,7 @@ public Topology topology() {
         this.purgeWorkerRunning(workerTaskResultKStream);
         this.detectNewWorker(workerInstanceKStream, workerTaskRunningKTable);
 
-        // build
-        Topology topology = builder.build();
-
-        if (log.isTraceEnabled()) {
-            log.trace(topology.describe().toString());
-        }
-
-        return topology;
+        return builder;
     }
 
     private void executionToExecutor(StreamsBuilder builder) {
@@ -940,7 +931,14 @@ public void run() {
         properties.put(StreamsConfig.DEFAULT_PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG, KafkaExecutorProductionExceptionHandler.class);
         properties.put(KafkaExecutorProductionExceptionHandler.APPLICATION_CONTEXT_CONFIG, applicationContext);
 
-        resultStream = kafkaStreamService.of(this.getClass(), this.topology(), properties);
+        // build
+        Topology topology = this.topology().build();
+
+        if (log.isTraceEnabled()) {
+            log.trace(topology.describe().toString());
+        }
+
+        resultStream = kafkaStreamService.of(this.getClass(), topology, properties);
         resultStream.start();
 
         applicationContext.registerSingleton(new KafkaTemplateExecutor(
diff --git a/runner-kafka/src/main/java/io/kestra/runner/kafka/services/KafkaStreamSourceService.java b/runner-kafka/src/main/java/io/kestra/runner/kafka/services/KafkaStreamSourceService.java
index 5227ad6fb5..87bf9918a3 100644
--- a/runner-kafka/src/main/java/io/kestra/runner/kafka/services/KafkaStreamSourceService.java
+++ b/runner-kafka/src/main/java/io/kestra/runner/kafka/services/KafkaStreamSourceService.java
@@ -1,5 +1,6 @@
 package io.kestra.runner.kafka.services;
 
+import io.kestra.core.services.TaskDefaultService;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.common.serialization.Serdes;
 import org.apache.kafka.common.utils.Bytes;
@@ -25,6 +26,9 @@ public class KafkaStreamSourceService {
     @Inject
     private KafkaAdminService kafkaAdminService;
 
+    @Inject
+    private TaskDefaultService taskDefaultService;
+
     public GlobalKTable<String, Flow> flowGlobalKTable(StreamsBuilder builder) {
         return builder
             .globalTable(
@@ -84,7 +88,11 @@ public KStream<String, KafkaExecutor.ExecutionWithFlow> withFlow(GlobalKTable<St
             .join(
                 flowGlobalKTable,
                 (key, value) -> Flow.uid(value),
-                (execution, flow) -> new KafkaExecutor.ExecutionWithFlow(flow, execution),
+                (execution, flow) -> {
+                    Flow flowWithDefaults = taskDefaultService.injectDefaults(flow);
+
+                    return new KafkaExecutor.ExecutionWithFlow(flowWithDefaults, execution);
+                },
                 Named.as("withFlow-join")
             );
     }
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 0089f3ac64..6d8f4f2f2c 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
@@ -39,6 +39,8 @@ public class MemoryExecutor extends AbstractExecutor {
     private final QueueInterface<WorkerTaskResult> workerTaskResultQueue;
     private final QueueInterface<LogEntry> logQueue;
     private final FlowService flowService;
+    private final TaskDefaultService taskDefaultService;
+
     private static final MemoryMultipleConditionStorage multipleConditionStorage = new MemoryMultipleConditionStorage();
 
     private static final ConcurrentHashMap<String, ExecutionState> executions = new ConcurrentHashMap<>();
@@ -57,7 +59,7 @@ public MemoryExecutor(
         ConditionService conditionService,
         TaskDefaultService taskDefaultService
     ) {
-        super(runContextFactory, metricRegistry, conditionService, taskDefaultService);
+        super(runContextFactory, metricRegistry, conditionService);
 
         this.flowRepository = flowRepository;
         this.executionQueue = executionQueue;
@@ -66,6 +68,7 @@ public MemoryExecutor(
         this.logQueue = logQueue;
         this.flowService = flowService;
         this.conditionService = conditionService;
+        this.taskDefaultService = taskDefaultService;
     }
 
     @Override
@@ -87,20 +90,20 @@ private void handleExecution(ExecutionState state) {
                 log.debug("Execution in with {}: {}", state.execution.toCrc32State(), state.execution.toStringState());
             }
 
-            Flow flow = this.flowRepository.findByExecution(state.execution);
+            Flow flow = taskDefaultService.injectDefaults(this.flowRepository.findByExecution(state.execution));
 
             Execution execution = state.execution;
 
             Optional<Execution> main = this.doMain(execution, flow);
             if (main.isPresent()) {
-                this.toExecution(main.get());
+                this.toExecution(main.get(), flow);
                 return;
             }
 
             try {
                 Optional<List<TaskRun>> nexts = this.doNexts(execution, flow);
                 if (nexts.isPresent() && deduplicateNexts(execution, nexts.get())) {
-                    this.toExecution(this.onNexts(flow, execution, nexts.get()));
+                    this.toExecution(this.onNexts(flow, execution, nexts.get()), flow);
                     return;
                 }
 
@@ -135,7 +138,7 @@ private void handleExecution(ExecutionState state) {
                     return;
                 }
             } catch (Exception e) {
-                handleFailedExecutionFromExecutor(execution, e);
+                handleFailedExecutionFromExecutor(execution, flow, e);
                 return;
             }
 
@@ -164,12 +167,12 @@ private void handleExecution(ExecutionState state) {
         }
     }
 
-    private void handleFailedExecutionFromExecutor(Execution execution, Exception e) {
+    private void handleFailedExecutionFromExecutor(Execution execution, Flow flow, Exception e) {
         Execution.FailedExecutionWithLog failedExecutionWithLog = execution.failedExecutionFromExecutor(e);
         try {
             failedExecutionWithLog.getLogs().forEach(logQueue::emit);
 
-            this.toExecution(failedExecutionWithLog.getExecution());
+            this.toExecution(failedExecutionWithLog.getExecution(), flow);
         } catch (Exception ex) {
             log.error("Failed to produce {}", e.getMessage(), ex);
         }
@@ -188,9 +191,7 @@ private ExecutionState saveExecution(Execution execution) {
         return queued;
     }
 
-    private void toExecution(Execution execution) {
-        Flow flow = this.flowRepository.findByExecution(execution);
-
+    private void toExecution(Execution execution, Flow flow) {
         if (log.isDebugEnabled()) {
             log.debug("Execution out with {}: {}", execution.toCrc32State(), execution.toStringState());
         }
@@ -234,13 +235,10 @@ private void workerTaskResultQueue(WorkerTaskResult message) {
                 }
             });
 
-            this.toExecution(executions.get(message.getTaskRun().getExecutionId()).execution);
-        }
-    }
-
-    private void handleFlowTrigger(Execution execution) {
-
+            Flow flow = taskDefaultService.injectDefaults(this.flowRepository.findByExecution(executions.get(message.getTaskRun().getExecutionId()).execution));
 
+            this.toExecution(executions.get(message.getTaskRun().getExecutionId()).execution, flow);
+        }
     }
 
     private boolean deduplicateWorkerTask(Execution execution, TaskRun taskRun) {


Test patch:
diff --git a/core/src/test/java/io/kestra/core/runners/RunContextTest.java b/core/src/test/java/io/kestra/core/runners/RunContextTest.java
index cb97a1c9f2..f870f4b9a9 100644
--- a/core/src/test/java/io/kestra/core/runners/RunContextTest.java
+++ b/core/src/test/java/io/kestra/core/runners/RunContextTest.java
@@ -1,25 +1,25 @@
 package io.kestra.core.runners;
 
-import org.exparity.hamcrest.date.ZonedDateTimeMatchers;
-import org.junit.jupiter.api.Test;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.executions.LogEntry;
-import io.kestra.core.models.executions.TaskRun;
 import io.kestra.core.models.executions.TaskRunAttempt;
 import io.kestra.core.models.executions.metrics.Counter;
 import io.kestra.core.models.executions.metrics.Timer;
 import io.kestra.core.queues.QueueFactoryInterface;
 import io.kestra.core.queues.QueueInterface;
 import io.kestra.core.utils.TestsUtils;
+import org.exparity.hamcrest.date.ZonedDateTimeMatchers;
+import org.junit.jupiter.api.Test;
 import org.slf4j.event.Level;
 
+import java.io.IOException;
+import java.net.URISyntaxException;
 import java.time.ZonedDateTime;
 import java.time.temporal.ChronoUnit;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Objects;
 import java.util.concurrent.TimeoutException;
-import java.util.stream.Collectors;
-
 import javax.inject.Inject;
 import javax.inject.Named;
 
@@ -33,6 +33,9 @@ class RunContextTest extends AbstractMemoryRunnerTest {
     @Named(QueueFactoryInterface.WORKERTASKLOG_NAMED)
     QueueInterface<LogEntry> workerTaskLogQueue;
 
+    @Inject
+    TaskDefaultsCaseTest taskDefaultsCaseTest;
+
     @Test
     void logs() throws TimeoutException {
         List<LogEntry> logs = new ArrayList<>();
@@ -91,4 +94,10 @@ void metrics() throws TimeoutException {
         assertThat(duration.getValue().getNano(), is(greaterThan(0)));
         assertThat(duration.getTags().get("format"), is("{{task.id}}"));
     }
+
+    @Test
+    void taskDefaults() throws TimeoutException, IOException, URISyntaxException {
+        repositoryLoader.load(Objects.requireNonNull(ListenersTest.class.getClassLoader().getResource("flows/tests/task-defaults.yaml")));
+        taskDefaultsCaseTest.taskDefaults();
+    }
 }
diff --git a/core/src/test/java/io/kestra/core/runners/TaskDefaultsCaseTest.java b/core/src/test/java/io/kestra/core/runners/TaskDefaultsCaseTest.java
new file mode 100644
index 0000000000..92c3d43ec8
--- /dev/null
+++ b/core/src/test/java/io/kestra/core/runners/TaskDefaultsCaseTest.java
@@ -0,0 +1,143 @@
+package io.kestra.core.runners;
+
+import io.kestra.core.exceptions.IllegalVariableEvaluationException;
+import io.kestra.core.models.executions.Execution;
+import io.kestra.core.models.executions.NextTaskRun;
+import io.kestra.core.models.executions.TaskRun;
+import io.kestra.core.models.hierarchies.GraphCluster;
+import io.kestra.core.models.hierarchies.RelationType;
+import io.kestra.core.models.tasks.FlowableTask;
+import io.kestra.core.models.tasks.ResolvedTask;
+import io.kestra.core.models.tasks.Task;
+import io.kestra.core.services.GraphService;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.ToString;
+import lombok.experimental.SuperBuilder;
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.concurrent.TimeoutException;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import javax.inject.Inject;
+import javax.inject.Singleton;
+import javax.validation.Valid;
+import javax.validation.constraints.NotEmpty;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.hasSize;
+import static org.hamcrest.Matchers.is;
+
+@Singleton
+public class TaskDefaultsCaseTest {
+    @Inject
+    private RunnerUtils runnerUtils;
+
+    public void taskDefaults() throws TimeoutException {
+        Execution execution = runnerUtils.runOne("io.kestra.tests", "task-defaults", Duration.ofSeconds(60));
+
+        assertThat(execution.getTaskRunList(), hasSize(8));
+
+        assertThat(execution.getTaskRunList().get(0).getTaskId(), is("first"));
+        assertThat(execution.getTaskRunList().get(0).getOutputs().get("def"), is("1"));
+        assertThat(execution.getTaskRunList().get(1).getTaskId(), is("second"));
+        assertThat(execution.getTaskRunList().get(1).getOutputs().get("def"), is("2"));
+        assertThat(execution.getTaskRunList().get(2).getTaskId(), is("third"));
+        assertThat(execution.getTaskRunList().get(2).getOutputs().get("def"), is("3"));
+
+        assertThat(execution.getTaskRunList().get(4).getTaskId(), is("err-first"));
+        assertThat(execution.getTaskRunList().get(4).getOutputs().get("def"), is("1"));
+        assertThat(execution.getTaskRunList().get(5).getTaskId(), is("err-second"));
+        assertThat(execution.getTaskRunList().get(5).getOutputs().get("def"), is("2"));
+        assertThat(execution.getTaskRunList().get(6).getTaskId(), is("err-third"));
+        assertThat(execution.getTaskRunList().get(6).getOutputs().get("def"), is("3"));
+    }
+
+    @SuperBuilder
+    @ToString
+    @EqualsAndHashCode
+    @Getter
+    @NoArgsConstructor
+    public static class DefaultSequential1 extends Task implements FlowableTask<DefaultSequential1.Output> {
+        @Valid
+        protected List<Task> errors;
+
+        @Valid
+        @NotEmpty
+        private List<Task> tasks;
+
+        private String def;
+
+        @Override
+        public GraphCluster tasksTree(Execution execution, TaskRun taskRun, List<String> parentValues) throws IllegalVariableEvaluationException {
+            GraphCluster subGraph = new GraphCluster(this, taskRun, parentValues, RelationType.SEQUENTIAL);
+
+            GraphService.sequential(
+                subGraph,
+                this.tasks,
+                this.errors,
+                taskRun,
+                execution
+            );
+
+            return subGraph;
+        }
+
+        @Override
+        public List<Task> allChildTasks() {
+            return Stream
+                .concat(
+                    this.tasks != null ? this.tasks.stream() : Stream.empty(),
+                    this.errors != null ? this.errors.stream() : Stream.empty()
+                )
+                .collect(Collectors.toList());
+        }
+
+        @Override
+        public List<ResolvedTask> childTasks(RunContext runContext, TaskRun parentTaskRun) throws IllegalVariableEvaluationException {
+            return FlowableUtils.resolveTasks(this.tasks, parentTaskRun);
+        }
+
+        @Override
+        public List<NextTaskRun> resolveNexts(RunContext runContext, Execution execution, TaskRun parentTaskRun) throws IllegalVariableEvaluationException {
+            return FlowableUtils.resolveSequentialNexts(
+                execution,
+                this.childTasks(runContext, parentTaskRun),
+                FlowableUtils.resolveTasks(this.errors, parentTaskRun),
+                parentTaskRun
+            );
+        }
+
+        @Override
+        public Output outputs(RunContext runContext, Execution execution, TaskRun parentTaskRun) throws IllegalVariableEvaluationException {
+            return Output.builder()
+                .def(this.def)
+                .build();
+        }
+
+        @SuperBuilder
+        @Getter
+        public static class Output implements io.kestra.core.models.tasks.Output {
+            private final String def;
+        }
+    }
+
+    @SuperBuilder
+    @ToString
+    @EqualsAndHashCode
+    @Getter
+    @NoArgsConstructor
+    public static class DefaultSequential2 extends DefaultSequential1 {
+    }
+
+    @SuperBuilder
+    @ToString
+    @EqualsAndHashCode
+    @Getter
+    @NoArgsConstructor
+    public static class DefaultSequential3 extends DefaultSequential1 {
+    }
+}
diff --git a/core/src/test/java/io/kestra/core/services/TaskDefaultServiceTest.java b/core/src/test/java/io/kestra/core/services/TaskDefaultServiceTest.java
index 3b6b9c75be..ef6460c8a7 100644
--- a/core/src/test/java/io/kestra/core/services/TaskDefaultServiceTest.java
+++ b/core/src/test/java/io/kestra/core/services/TaskDefaultServiceTest.java
@@ -1,7 +1,12 @@
 package io.kestra.core.services;
 
 import com.google.common.collect.ImmutableMap;
-import io.micronaut.context.ApplicationContext;
+import io.kestra.core.models.flows.Flow;
+import io.kestra.core.models.flows.TaskDefault;
+import io.kestra.core.models.tasks.RunnableTask;
+import io.kestra.core.models.tasks.Task;
+import io.kestra.core.models.tasks.VoidOutput;
+import io.kestra.core.runners.RunContext;
 import io.micronaut.test.extensions.junit5.annotation.MicronautTest;
 import lombok.EqualsAndHashCode;
 import lombok.Getter;
@@ -9,12 +14,6 @@
 import lombok.ToString;
 import lombok.experimental.SuperBuilder;
 import org.junit.jupiter.api.Test;
-import io.kestra.core.models.flows.Flow;
-import io.kestra.core.models.flows.TaskDefault;
-import io.kestra.core.models.tasks.RunnableTask;
-import io.kestra.core.models.tasks.Task;
-import io.kestra.core.models.tasks.VoidOutput;
-import io.kestra.core.runners.RunContext;
 
 import java.util.Collections;
 import java.util.List;
@@ -30,9 +29,6 @@ class TaskDefaultServiceTest {
     @Inject
     private TaskDefaultService taskDefaultService;
 
-    @Inject
-    private ApplicationContext applicationContext;
-
     @Test
     public void injectFlowAndGlobals() {
         DefaultTester task = DefaultTester.builder()
@@ -50,17 +46,17 @@ public void injectFlowAndGlobals() {
             ))))
             .build();
 
-        DefaultTester injected = taskDefaultService.injectDefaults(task, flow);
-
-        assertThat(injected.getValue(), is(1));
-        assertThat(injected.getSet(), is(666));
-        assertThat(injected.getDoubleValue(), is(19D));
-        assertThat(injected.getArrays().size(), is(2));
-        assertThat(injected.getArrays(), containsInAnyOrder(1, 2));
-        assertThat(injected.getProperty().getHere(), is("me"));
-        assertThat(injected.getProperty().getLists().size(), is(1));
-        assertThat(injected.getProperty().getLists().get(0).getVal().size(), is(1));
-        assertThat(injected.getProperty().getLists().get(0).getVal().get("key"), is("test"));
+        Flow injected = taskDefaultService.injectDefaults(flow);
+
+        assertThat(((DefaultTester) injected.getTasks().get(0)).getValue(), is(1));
+        assertThat(((DefaultTester) injected.getTasks().get(0)).getSet(), is(666));
+        assertThat(((DefaultTester) injected.getTasks().get(0)).getDoubleValue(), is(19D));
+        assertThat(((DefaultTester) injected.getTasks().get(0)).getArrays().size(), is(2));
+        assertThat(((DefaultTester) injected.getTasks().get(0)).getArrays(), containsInAnyOrder(1, 2));
+        assertThat(((DefaultTester) injected.getTasks().get(0)).getProperty().getHere(), is("me"));
+        assertThat(((DefaultTester) injected.getTasks().get(0)).getProperty().getLists().size(), is(1));
+        assertThat(((DefaultTester) injected.getTasks().get(0)).getProperty().getLists().get(0).getVal().size(), is(1));
+        assertThat(((DefaultTester) injected.getTasks().get(0)).getProperty().getLists().get(0).getVal().get("key"), is("test"));
     }
 
     @SuperBuilder
diff --git a/core/src/test/resources/flows/tests/task-defaults.yaml b/core/src/test/resources/flows/tests/task-defaults.yaml
new file mode 100644
index 0000000000..aad35c44bd
--- /dev/null
+++ b/core/src/test/resources/flows/tests/task-defaults.yaml
@@ -0,0 +1,47 @@
+id: task-defaults
+namespace: io.kestra.tests
+
+taskDefaults:
+  - type: io.kestra.core.runners.TaskDefaultsCaseTest$DefaultSequential1
+    values:
+      def: "1"
+
+  - type: io.kestra.core.runners.TaskDefaultsCaseTest$DefaultSequential2
+    values:
+      def: "2"
+
+  - type: io.kestra.core.runners.TaskDefaultsCaseTest$DefaultSequential3
+    values:
+      def: "3"
+
+tasks:
+  - id: first
+    type: io.kestra.core.runners.TaskDefaultsCaseTest$DefaultSequential1
+    tasks:
+    - id: second
+      type: io.kestra.core.runners.TaskDefaultsCaseTest$DefaultSequential2
+      tasks:
+        - id: third
+          type: io.kestra.core.runners.TaskDefaultsCaseTest$DefaultSequential3
+          tasks:
+          - id: ko
+            type: io.kestra.core.tasks.scripts.Bash
+            commands:
+              - 'exit 1'
+  - id: 2_end
+    type: io.kestra.core.tasks.debugs.Return
+    format: "{{task.id}} > {{taskrun.startDate}}"
+
+errors:
+  - id: err-first
+    type: io.kestra.core.runners.TaskDefaultsCaseTest$DefaultSequential1
+    tasks:
+      - id: err-second
+        type: io.kestra.core.runners.TaskDefaultsCaseTest$DefaultSequential2
+        tasks:
+          - id: err-third
+            type: io.kestra.core.runners.TaskDefaultsCaseTest$DefaultSequential3
+            tasks:
+            - id: end
+              type: io.kestra.core.tasks.debugs.Return
+              format: "{{task.id}} > {{taskrun.startDate}}"
\ No newline at end of file
diff --git a/runner-kafka/src/test/java/io/kestra/runner/kafka/KafkaExecutorTest.java b/runner-kafka/src/test/java/io/kestra/runner/kafka/KafkaExecutorTest.java
index 8868062e30..ad569e989b 100644
--- a/runner-kafka/src/test/java/io/kestra/runner/kafka/KafkaExecutorTest.java
+++ b/runner-kafka/src/test/java/io/kestra/runner/kafka/KafkaExecutorTest.java
@@ -75,7 +75,7 @@ void init() throws IOException, URISyntaxException {
         properties.put(StreamsConfig.APPLICATION_ID_CONFIG, "unit-test");
         properties.put(StreamsConfig.STATE_DIR_CONFIG, "/tmp/kafka-stream-unit/" + UUID.randomUUID());
 
-        testTopology = new TopologyTestDriver(stream.topology(), properties);
+        testTopology = new TopologyTestDriver(stream.topology().build(), properties);
 
         applicationContext.registerSingleton(new KafkaTemplateExecutor(
             testTopology.getKeyValueStore("template")
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 3e1cc50da7..d952c1b466 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
@@ -1,8 +1,6 @@
 package io.kestra.runner.kafka;
 
 import com.google.common.collect.ImmutableMap;
-import org.apache.kafka.common.errors.RecordTooLargeException;
-import org.junit.jupiter.api.Test;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.executions.LogEntry;
 import io.kestra.core.models.flows.State;
@@ -14,6 +12,8 @@
 import io.kestra.core.tasks.flows.EachSequentialTest;
 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.Test;
 
 import java.io.IOException;
 import java.net.URISyntaxException;
@@ -39,6 +39,9 @@ class KafkaRunnerTest extends AbstractKafkaRunnerTest {
     @Inject
     private MultipleConditionTriggerCaseTest multipleConditionTriggerCaseTest;
 
+    @Inject
+    private TaskDefaultsCaseTest taskDefaultsCaseTest;
+
     @Inject
     private TemplateRepositoryInterface templateRepository;
 
@@ -235,4 +238,10 @@ void eachWithNull() throws Exception {
     void withTemplate() throws Exception {
         TemplateTest.withTemplate(runnerUtils, templateRepository);
     }
+
+    @Test
+    void taskDefaults() throws TimeoutException, IOException, URISyntaxException {
+        repositoryLoader.load(Objects.requireNonNull(ListenersTest.class.getClassLoader().getResource("flows/tests/task-defaults.yaml")));
+        taskDefaultsCaseTest.taskDefaults();
+    }
 }