Repository: kestra-io/kestra (java)

Issue:
ForEach with unlimited concurrency executes the tasks in reverse order
### Describe the issue

Use the following flow:

```yaml
id: hummingbird_941521
namespace: company.team

tasks:

  - id: foreach
    type: io.kestra.plugin.core.flow.ForEach
    values: "{{ range(1, 16) }}"
    concurrencyLimit: 0
    tasks:
      - id: sleep
        type: io.kestra.plugin.core.flow.Sleep
        duration: PT1S
```

You can see that tasks are executed in reverse order (set worker threads to 2 to see it more easily)

![Image](https://github.com/user-attachments/assets/07894447-be8a-4079-861f-f14aa8121cb0)



### Environment

- Kestra Version: develop


Pull request description:
fix(plugins)*: ForEach executes task in the wrong order when no concu…
…rrency limit

Fixes #8904


Patch:
diff --git a/core/src/main/java/io/kestra/core/models/executions/Execution.java b/core/src/main/java/io/kestra/core/models/executions/Execution.java
index dde88c617..fdb9ceba0 100644
--- a/core/src/main/java/io/kestra/core/models/executions/Execution.java
+++ b/core/src/main/java/io/kestra/core/models/executions/Execution.java
@@ -853,10 +853,12 @@ public class Execution implements DeletedInterface, TenantInterface {
             .forEach((taskId, taskRuns) -> {
                 Map<String, Object> taskOutputs = new HashMap<>();
                 for (TaskRun current : taskRuns) {
-                    if (current.getIteration() != null) {
-                        taskOutputs = MapUtils.merge(taskOutputs, outputs(current, byIds));
-                    } else {
-                        taskOutputs.putAll(outputs(current, byIds));
+                    if (!MapUtils.isEmpty(current.getOutputs())) {
+                        if (current.getIteration() != null) {
+                            taskOutputs = MapUtils.merge(taskOutputs, outputs(current, byIds));
+                        } else {
+                            taskOutputs.putAll(outputs(current, byIds));
+                        }
                     }
                 }
                 result.put(taskId, taskOutputs);
diff --git a/core/src/main/java/io/kestra/core/runners/FlowableUtils.java b/core/src/main/java/io/kestra/core/runners/FlowableUtils.java
index 62b9c020f..3e9e81d4f 100644
--- a/core/src/main/java/io/kestra/core/runners/FlowableUtils.java
+++ b/core/src/main/java/io/kestra/core/runners/FlowableUtils.java
@@ -281,8 +281,7 @@ public class FlowableUtils {
             return collect.values().stream()
                 .limit(concurrencySlots)
                 .map(resolvedTasks -> resolvedTasks.getFirst().toNextTaskRun(execution))
-                .toList()
-                .reversed();
+                .toList();
         }
 
         // start as many tasks as we have concurrency slots
diff --git a/core/src/main/java/io/kestra/core/runners/RunVariables.java b/core/src/main/java/io/kestra/core/runners/RunVariables.java
index bbfe2eaad..0d70fd4a7 100644
--- a/core/src/main/java/io/kestra/core/runners/RunVariables.java
+++ b/core/src/main/java/io/kestra/core/runners/RunVariables.java
@@ -12,6 +12,7 @@ import io.kestra.core.models.flows.input.SecretInput;
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.models.triggers.AbstractTrigger;
 import io.kestra.core.utils.ListUtils;
+import io.kestra.core.utils.MapUtils;
 import lombok.AllArgsConstructor;
 import lombok.With;
 
@@ -170,6 +171,7 @@ public final class RunVariables {
             this.secretKey = secretKey;
         }
 
+        // Note: for performance reason, cloning maps should be avoided as much as possible.
         @Override
         public Map<String, Object> build(final RunContextLogger logger) {
             ImmutableMap.Builder<String, Object> builder = ImmutableMap.builder();
@@ -228,7 +230,7 @@ public final class RunVariables {
                 builder.put("execution", executionMap.build());
 
                 if (execution.getTaskRunList() != null) {
-                    Map<String, Object> outputs = new HashMap<>(execution.outputs());
+                    Map<String, Object> outputs = execution.outputs();
                     if (decryptVariables) {
                         final Secret secret = new Secret(secretKey, logger);
                         outputs = secret.decrypt(outputs);
@@ -244,11 +246,13 @@ public final class RunVariables {
                             } else {
                                 if (tasksMap.containsKey(taskRun.getTaskId())) {
                                     @SuppressWarnings("unchecked")
-                                    Map<String, Object> taskRunMap = new HashMap<>((Map<String, Object>) tasksMap.get(taskRun.getTaskId()));
+                                    Map<String, Object> taskRunMap = (Map<String, Object>) tasksMap.get(taskRun.getTaskId());
                                     taskRunMap.put(taskRun.getValue(), Map.of("state", taskRun.getState().getCurrent()));
                                     tasksMap.put(taskRun.getTaskId(), taskRunMap);
                                 } else {
-                                    tasksMap.put(taskRun.getTaskId(), Map.of(taskRun.getValue(), Map.of("state", taskRun.getState().getCurrent())));
+                                    Map<String, Object> taskRunMap = new HashMap<>();
+                                    taskRunMap.put(taskRun.getValue(), Map.of("state", taskRun.getState().getCurrent()));
+                                    tasksMap.put(taskRun.getTaskId(), taskRunMap);
                                 }
                             }
                         }
@@ -318,7 +322,7 @@ public final class RunVariables {
 
             // Kestra configuration
             if (kestraConfiguration != null) {
-                Map<String, String> kestra = new HashMap<>();
+                Map<String, String> kestra = HashMap.newHashMap(2);
                 if (kestraConfiguration.environment() != null) {
                     kestra.put("environment", kestraConfiguration.environment());
                 }
diff --git a/core/src/main/java/io/kestra/core/utils/MapUtils.java b/core/src/main/java/io/kestra/core/utils/MapUtils.java
index 10b05cf6a..0e89de7ba 100644
--- a/core/src/main/java/io/kestra/core/utils/MapUtils.java
+++ b/core/src/main/java/io/kestra/core/utils/MapUtils.java
@@ -5,7 +5,7 @@ import lombok.extern.slf4j.Slf4j;
 
 import java.util.*;
 
-@SuppressWarnings({"rawtypes", "unchecked"})
+@SuppressWarnings({"unchecked"})
 @Slf4j
 public class MapUtils {
     private static final String CONFLICT_AT_KEY_MSG = "Conflict at key: '{}', ignoring it. Map keys are: {}";
@@ -23,86 +23,61 @@ public class MapUtils {
             return a;
         }
 
-        Map copy = copyMap(a);
-
-        Map<String, Object> copyMap = b
-            .entrySet()
-            .stream()
-            .collect(
-                () -> HashMap.newHashMap(copy.size()),
-                (m, v) -> {
-                    Object original = copy.get(v.getKey());
-                    Object value = v.getValue();
-                    Object found;
-
-                    if (value == null) {
-                        found = original;
-                    } else if (original == null) {
-                        found = value;
-                    } else if (value instanceof Map mapValue && original instanceof Map mapOriginal) {
-                        found = merge(mapOriginal, mapValue);
-                    } else if (value instanceof Collection collectionValue
-                        && original instanceof Collection collectionOriginal) {
-                        found = mergeCollections(collectionOriginal, collectionValue);
-                    } else {
-                        found = value;
-                    }
-
-                    m.put(v.getKey(), found);
-                },
-                HashMap::putAll
-            );
-
-        copy.putAll(copyMap);
-
-        return copy;
-    }
+        Map<String, Object> result = HashMap.newHashMap(Math.max(a.size(), b.size()));
+        result.putAll(deepCloneMap(a));
+
+        for (Map.Entry<String, Object> entry : b.entrySet()) {
+            String key = entry.getKey();
+            Object valueB = entry.getValue();
+            Object valueA = result.get(key);
+
+            Object mergedValue;
+            if (valueB == null) {
+                mergedValue = valueA;
+            } else if (valueA == null) {
+                mergedValue = valueB;
+            } else if (valueA instanceof Map<?, ?> mapA && valueB instanceof Map<?, ?> mapB) {
+                mergedValue = merge(castMap(mapA), castMap(mapB));
+            } else if (valueA instanceof Collection<?> colA && valueB instanceof Collection<?> colB) {
+                mergedValue = mergeCollections(colA, colB);
+            } else {
+                mergedValue = valueB;
+            }
+
+            result.put(key, mergedValue);
+        }
 
-    private static Collection mergeCollections(Collection collectionOriginal, Collection collectionValue) {
-        List<?> newList = new ArrayList<>(collectionOriginal.size() + collectionValue.size());
-        newList.addAll(collectionOriginal);
-        newList.addAll(collectionValue);
-        return newList;
+        return result;
     }
 
-    private static Map copyMap(Map original) {
-        return ((Map<?, ?>) original)
-            .entrySet()
-            .stream()
-            .collect(
-                () -> HashMap.newHashMap(original.size()),
-                (map, entry) -> {
-                    Object value = entry.getValue();
-                    Object found;
-
-                    if (value instanceof Map mapValue) {
-                        found = cloneMap(mapValue);
-                    } else if (value instanceof Collection collectionValue) {
-                        found = cloneCollection(collectionValue);
-                    } else {
-                        found = value;
-                    }
-
-                    map.put(entry.getKey(), found);
-
-                },
-                Map::putAll
-            );
+    private static Map<String, Object> deepCloneMap(Map<String, Object> original) {
+        Map<String, Object> cloned = new HashMap<>(original.size());
+        for (Map.Entry<String, Object> entry : original.entrySet()) {
+            cloned.put(entry.getKey(), deepClone(entry.getValue()));
+        }
+        return cloned;
     }
 
-    private static Map cloneMap(Map elements) {
-        try {
-            Map newInstance = elements.getClass().getDeclaredConstructor().newInstance();
-            newInstance.putAll(elements);
-            return newInstance;
-        } catch (Exception e) {
-            return new HashMap(elements);
+    private static Object deepClone(Object value) {
+        if (value instanceof Map<?, ?> map) {
+            return deepCloneMap(castMap(map));
+        } else if (value instanceof Collection<?> col) {
+            return cloneCollection(col);
+        } else {
+            return value;
         }
     }
 
-    private static Collection cloneCollection(Collection elements) {
+    private static Collection<?> mergeCollections(Collection<?> colA, Collection<?> colB) {
+        List<Object> merged = new ArrayList<>(colA.size() + colB.size());
+        merged.addAll(colA);
+        merged.addAll(colB);
+        return merged;
+    }
+
+    private static Collection<?> cloneCollection(Collection<?> elements) {
         try {
-            Collection newInstance = elements.getClass().getDeclaredConstructor().newInstance();
+            Collection<Object> newInstance = elements.getClass().getDeclaredConstructor().newInstance();
             newInstance.addAll(elements);
             return newInstance;
         } catch (Exception e) {
@@ -110,6 +85,11 @@ public class MapUtils {
         }
     }
 
+    @SuppressWarnings("unchecked")
+    private static Map<String, Object> castMap(Map<?, ?> map) {
+        return (Map<String, Object>) map;
+    }
+
     /**
      * Utility method for merging multiple {@link Map}s that can contains nullable values.
      * Note that the maps provided are assumed to be flat, so this method does not perform a recursive merge.
diff --git a/jmh-benchmarks/README.md b/jmh-benchmarks/README.md
new file mode 100644
index 000000000..34ea45156
--- /dev/null
+++ b/jmh-benchmarks/README.md
@@ -0,0 +1,17 @@
+# JMH-Benchmarks module
+
+This module contains benchmarks written using JMH from OpenJDK.
+
+## Running Benchmarks
+
+**To run all benchmarks**
+
+```bash
+./gradlew jmh:jmh
+```
+
+**To run a specific benchmark**
+
+```bash
+./gradlew jmh:jmh -Pjmh.include=io.kestra.core.utils.MapUtilsBenchmark
+```
\ No newline at end of file
diff --git a/jmh-benchmarks/build.gradle b/jmh-benchmarks/build.gradle
new file mode 100644
index 000000000..c958df366
--- /dev/null
+++ b/jmh-benchmarks/build.gradle
@@ -0,0 +1,22 @@
+plugins {
+    id "me.champeau.jmh" version "0.7.2"
+}
+
+configurations {
+    tests
+    implementation.extendsFrom(micronaut)
+}
+
+tasks.register('copyGradleProperties', Copy) {
+    group = "build"
+    shouldRunAfter compileJava
+
+    from '../gradle.properties'
+    into 'src/main/resources'
+}
+
+processResources.dependsOn copyGradleProperties
+
+dependencies {
+    jmh project(':core')
+}
diff --git a/jmh-benchmarks/src/jmh/java/io/kestra/core/utils/MapUtilsBenchmark.java b/jmh-benchmarks/src/jmh/java/io/kestra/core/utils/MapUtilsBenchmark.java
new file mode 100644
index 000000000..e8269246a
--- /dev/null
+++ b/jmh-benchmarks/src/jmh/java/io/kestra/core/utils/MapUtilsBenchmark.java
@@ -0,0 +1,35 @@
+package io.kestra.core.utils;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import org.openjdk.jmh.annotations.*;
+
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.MICROSECONDS)
+@State(Scope.Thread)
+public class MapUtilsBenchmark {
+    private Map<String, Object> mapA;
+    private Map<String, Object> mapB;
+
+    @Setup(Level.Invocation)
+    public void setup() {
+        mapA = new HashMap<>();
+        mapB = new HashMap<>();
+        for (int i = 0; i < 100; i++) {
+            mapA.put("key" + i, "valueA" + i);
+            mapB.put("key" + i, "valueB" + i);
+        }
+        mapA.put("nested", Map.of("a", 1));
+        mapB.put("nested", Map.of("b", 2));
+    }
+
+    /**
+     * @see <a href="https://github.com/kestra-io/kestra/pull/8914">KESTRA#8914</a>
+     */
+    @Benchmark
+    public Map<String, Object> testMerge() {
+        return MapUtils.merge(mapA, mapB);
+    }
+}
diff --git a/settings.gradle b/settings.gradle
index 69140b423..430d967e1 100644
--- a/settings.gradle
+++ b/settings.gradle
@@ -1,4 +1,4 @@
-rootProject.name="kestra"
+rootProject.name = "kestra"
 
 include 'platform'
 
@@ -24,3 +24,4 @@ include 'processor'
 include 'script'
 include 'e2e-tests'
 
+include 'jmh-benchmarks'
\ No newline at end of file
diff --git a/ui/src/components/code/components/Save.vue b/ui/src/components/code/components/Save.vue
index b924af91c..8430eb45a 100644
--- a/ui/src/components/code/components/Save.vue
+++ b/ui/src/components/code/components/Save.vue
@@ -14,7 +14,7 @@
     const props = defineProps({
         what: {
             type: String,
-            required: true,
+            required: false,
             default: "tasks"
         },
     });
diff --git a/ui/src/components/code/segments/Task.vue b/ui/src/components/code/segments/Task.vue
index cf0b0e2f7..e14c5c6b6 100644
--- a/ui/src/components/code/segments/Task.vue
+++ b/ui/src/components/code/segments/Task.vue
@@ -19,6 +19,7 @@
         <ValidationError v-if="false" :errors link />
 
         <Save
+            v-if="!lastBreadcrumb.component"
             :disabled="(errors?.length ?? 0) > 0"
             @click="() => {
                 saveTask();
diff --git a/ui/src/components/flows/MetadataInputsContent.vue b/ui/src/components/flows/MetadataInputsContent.vue
index 0812a83b0..535c508a3 100644
--- a/ui/src/components/flows/MetadataInputsContent.vue
+++ b/ui/src/components/flows/MetadataInputsContent.vue
@@ -19,6 +19,7 @@
         :schema="inputSchema?.schema?.properties"
         :properties="inputSchema?.schema?.properties?.properties"
         :definitions="inputSchema?.schema?.definitions"
+        metadata-inputs
     />
 
     <Save @click="update" what="input" class="w-100 mt-3" />
@@ -30,7 +31,7 @@
 </script>
 
 <script>
-    import {mapState} from "vuex";
+    import {mapState, mapGetters} from "vuex";
     import {BREADCRUMB_INJECTION_KEY, PANEL_INJECTION_KEY} from "../code/injectionKeys";
 
     export default {
@@ -51,6 +52,7 @@
         },
         computed: {
             ...mapState("plugin", ["inputSchema", "inputsType"]),
+            ...mapGetters("flow", ["flowYamlMetadata"]),
         },
         created() {
             if (this.inputs && this.inputs.length > 0) {
@@ -105,7 +107,7 @@
                 } else {
                     this.panel = undefined;
                     this.breadcrumbs.pop();
-                    this.$emit("update:modelValue", [...this.newInputs.filter(v => v.id)]);
+                    this.$emit("update:modelValue", [this.flowYamlMetadata?.inputs]);
                 }
             },
             updateSelected(value) {
diff --git a/ui/src/components/flows/tasks/TaskComplex.vue b/ui/src/components/flows/tasks/TaskComplex.vue
index dbedb72bc..38362d84f 100644
--- a/ui/src/components/flows/tasks/TaskComplex.vue
+++ b/ui/src/components/flows/tasks/TaskComplex.vue
@@ -26,11 +26,14 @@
             panel: {from: PANEL_INJECTION_KEY},
             breadcrumbs: {from: BREADCRUMB_INJECTION_KEY},
         },
+        props:{
+            metadataInputs: {type: Boolean, default: false},
+        },
         methods: {
             openPanel() {
                 const current = {...this.panel};
 
-                this.panel = h(TaskComplexContent, {
+                const component = h(TaskComplexContent, {
                     modelValue: this.modelValue,
                     schema: this.schema,
                     definitions: this.definitions,
@@ -38,9 +41,11 @@
                     root: this.root,
                     "onUpdate:modelValue": this.onInput,
                     previousPanel: current,
+                    metadataInputs: this.metadataInputs,
                 });
 
-                this.breadcrumbs.push({label: this.root});
+                this.panel = component;
+                this.breadcrumbs.push({label: this.root, component});
             },
         },
     };
diff --git a/ui/src/components/flows/tasks/TaskComplexContent.vue b/ui/src/components/flows/tasks/TaskComplexContent.vue
index b37dd40b4..902dc6f39 100644
--- a/ui/src/components/flows/tasks/TaskComplexContent.vue
+++ b/ui/src/components/flows/tasks/TaskComplexContent.vue
@@ -3,22 +3,26 @@
         :model-value
         :schema
         :definitions
-        @update:model-value="(e) => update(e)"
-        :properties="definitions[Object.keys(definitions)?.[0]]?.properties"
+        @update:model-value="(v) => update(v)"
+        :properties="computedProperties"
         expand-optional
     />
-    <Save @click="closePanel" what="input" class="w-100 mt-3" />
+    <Save @click="closePanel" class="w-100 mt-3" />
 </template>
 
 <script>
     import {h} from "vue";
+
     import Task from "./Task";
+
     import TaskObject from "./TaskObject.vue";
     import Save from "../../code/components/Save.vue";
+
     import {
         BREADCRUMB_INJECTION_KEY,
         PANEL_INJECTION_KEY,
     } from "../../code/injectionKeys";
+
     export default {
         mixins: [Task],
         inject: {
@@ -26,12 +30,29 @@
             breadcrumbs: {from: BREADCRUMB_INJECTION_KEY},
         },
         props: {
+            metadataInputs: {type: Boolean, default: false},
             previousPanel: {type: Object, default: () => ({})},
         },
         components: {TaskObject, Save},
+        data() {
+            return {
+                currentTask: {},
+            };
+        },
+        computed: {
+            computedProperties() {
+                const type = this.schema.$ref.split("/").pop();
+                return this.definitions[type]?.properties;
+            },
+        },
         methods: {
-            update(e) {
-                this.onInput({...this.task, [this.root]: e});
+            update(v) {
+                const updated = {
+                    ...this.task,
+                    [this.root]: {...this.currentTask[this.root], ...v},
+                };
+                this.currentTask = updated;
+                this.onInput(this.metadataInputs ? updated : v);
             },
             closePanel() {
                 this.panel = this.previousPanel ? h(this.previousPanel) : null;
@@ -39,4 +60,4 @@
             },
         },
     };
-</script>
\ No newline at end of file
+</script>
diff --git a/ui/src/components/flows/tasks/TaskObject.vue b/ui/src/components/flows/tasks/TaskObject.vue
index a7833a480..1cbf5e8d1 100644
--- a/ui/src/components/flows/tasks/TaskObject.vue
+++ b/ui/src/components/flows/tasks/TaskObject.vue
@@ -43,6 +43,7 @@
                                     :required="isRequired(key)"
                                     :definitions="definitions"
                                     class="mt-1 mb-2 wrapper"
+                                    v-bind="getType(schema, key) === 'complex' ? {metadataInputs} : {}"
                                 />
                             </el-form-item>
                         </template>
@@ -88,6 +89,7 @@
                             :required="isRequired(key)"
                             :definitions="definitions"
                             class="mt-1 mb-2 wrapper"
+                            v-bind="getType(schema, key) === 'complex' ? {metadataInputs} : {}"
                         />
                     </el-form-item>
                 </template>
@@ -137,6 +139,7 @@
                                             :required="isRequired(key)"
                                             :definitions="definitions"
                                             class="mt-1 mb-2 wrapper"
+                                            v-bind="getType(schema, key) === 'complex' ? {metadataInputs} : {}"
                                         />
                                     </el-form-item>
                                 </template>
@@ -182,6 +185,7 @@
                                     :required="isRequired(key)"
                                     :definitions="definitions"
                                     class="mt-1 mb-2 wrapper"
+                                    v-bind="getType(schema, key) === 'complex' ? {metadataInputs} : {}"
                                 />
                             </el-form-item>
                         </template>
@@ -233,6 +237,7 @@
                                             :required="isRequired(key)"
                                             :definitions="definitions"
                                             class="mt-1 mb-2 wrapper"
+                                            v-bind="getType(schema, key) === 'complex' ? {metadataInputs} : {}"
                                         />
                                     </el-form-item>
                                 </template>
@@ -278,6 +283,7 @@
                                     :required="isRequired(key)"
                                     :definitions="definitions"
                                     class="mt-1 mb-2 wrapper"
+                                    v-bind="getType(schema, key) === 'complex' ? {metadataInputs} : {}"
                                 />
                             </el-form-item>
                         </template>
@@ -369,7 +375,8 @@
                 type: Object,
                 default: () => ({}),
             },
-            expandOptional: {type: Boolean, default: false}
+            expandOptional: {type: Boolean, default: false},
+            metadataInputs: {type: Boolean, default: false}
         },
         emits: ["update:modelValue"],
         data() {
diff --git a/ui/src/components/inputs/EditorWrapper.vue b/ui/src/components/inputs/EditorWrapper.vue
index 42a29893f..87c4f2830 100644
--- a/ui/src/components/inputs/EditorWrapper.vue
+++ b/ui/src/components/inputs/EditorWrapper.vue
@@ -14,7 +14,9 @@
         @save="save"
         @execute="execute"
     >
-        <KeyShortcuts />
+        <template #absolute>
+            <KeyShortcuts v-if="isCurrentTabFlow" />
+        </template>
     </editor>
 </template>
 


Test patch:
diff --git a/cli/src/test/java/io/kestra/cli/services/FileChangedEventListenerTest.java b/cli/src/test/java/io/kestra/cli/services/FileChangedEventListenerTest.java
index 695fe04ca..d826e2c4e 100644
--- a/cli/src/test/java/io/kestra/cli/services/FileChangedEventListenerTest.java
+++ b/cli/src/test/java/io/kestra/cli/services/FileChangedEventListenerTest.java
@@ -44,6 +44,7 @@ class FileChangedEventListenerTest {
     @AfterAll
     static void tearDown() throws IOException {
         if (Files.exists(Path.of(FILE_WATCH))) {
+            FileUtils.cleanDirectory(Path.of(FILE_WATCH).toFile());
             FileUtils.deleteDirectory(Path.of(FILE_WATCH).toFile());
         }
     }
diff --git a/tests/src/main/java/io/kestra/core/utils/TestsUtils.java b/tests/src/main/java/io/kestra/core/utils/TestsUtils.java
index 24f883d8b..2d6b93c67 100644
--- a/tests/src/main/java/io/kestra/core/utils/TestsUtils.java
+++ b/tests/src/main/java/io/kestra/core/utils/TestsUtils.java
@@ -136,7 +136,7 @@ abstract public class TestsUtils {
             .withState(State.Type.RUNNING);
     }
 
-    private static TaskRun mockTaskRun(Execution execution, Task task) {
+    public static TaskRun mockTaskRun(Execution execution, Task task) {
         return TaskRun.builder()
             .id(IdUtils.create())
             .executionId(execution.getId())