Repository: kestra-io/kestra (java)

Issue:
Add a new `ForEach` core task to unify and simplify `EachParallel` and `EachSequential`
### Feature description

## Problem

Some users getting started with Kestra are confused as to why a list of tasks specified within the `EachParallel` task is not executed sequentially, one after the other, for each parallel group of tasks. Many find it confusing that they need to wrap these tasks into a Sequential task to achieve parallelism for an entire group of tasks. 

At the same time, the `EachSequential` task seems to be functionally equivalent to `EachParallel` using the `concurrent` property set to 1. 

Lastly, the naming of tasks and properties is confusing:
1. The name of the task is `EachParallel`, and then instead of `parallel`, it has a `concurrent` property. It would make more sense to have `EachConcurrent` in this case to be precise or change that property to `parallel`
2. Both of these tasks accept a list of `values`. However, the property name is configured as a singular `value`, which is a little confusing given that it accepts a list of `values`, not a single value 
3. The task is doing a For loop. Some users may search for such looping task in autocompletion using "for" and they may not find it. Using the name `ForEach` will be easy to find for users looking for "each" and "for" keywords.

## Solution

Before the 1.0 release, it's important to unify both tasks into a single `ForEach` task that:
1. Allows `sequential` and `concurrent` execution of tasks simply by configuring it as task property `concurrent`, not as a dedicated task
2. Uses `values` property in plural or `items` to unify with `ForEachItem`

Proposed syntax:

```yaml
id: pythonPartitionsMetrics
namespace: blueprint
description: Process partitions in parallel

tasks:
  - id: getPartitions
    type: io.kestra.plugin.scripts.python.Script
    runner: DOCKER
    docker:
      image: ghcr.io/kestra-io/pydata:latest
    script: |
      from kestra import Kestra
      partitions = [f"file_{nr}.parquet" for nr in range(1, 1000)]
      Kestra.outputs({'partitions': partitions})

  - id: processPartitions
    type: io.kestra.core.tasks.flows.ForEach
    items: '{{outputs.getPartitions.vars.partitions}}'
    concurrent: 10 # set to 1 to get EachSequential-like processing
    tasks:
      - id: log_start
        type: io.kestra.core.tasks.log.Log
        message: staring processing {{ taskrun.item }}

      - id: partition
        type: io.kestra.plugin.scripts.python.Script
        runner: DOCKER
        docker:
          image: ghcr.io/kestra-io/pydata:latest
        script: |
          import random
          import time
          from kestra import Kestra

          filename = '{{ taskrun.item }}'
          print(f"Reading and processing partition {filename}")
          nr_rows = random.randint(1, 1000)
          processing_time = random.randint(1, 20)
          time.sleep(processing_time)
          Kestra.counter('nr_rows', nr_rows, {'partition': filename})
          Kestra.timer('processing_time', processing_time, {'partition': filename})

      - id: log_end
        type: io.kestra.core.tasks.log.Log
        message: finished processing {{ taskrun.item }}
```

### Allowing failure of parallel child tasks

This new task should make it possible for all child tasks to run till completion in a non-blocking way even if some child task fails. We were considering `continueOnError: boolean`. However, if `allowFailure` will be added to the core tasks, there seems to be no need for the extra `continueOnError` -- see  https://github.com/kestra-io/kestra/issues/2248.



Pull request:
feat(core): provide a ForEach task
Fixes #2137

Patch:
diff --git a/build.gradle b/build.gradle
index 2cab55f0c..1d7d8d673 100644
--- a/build.gradle
+++ b/build.gradle
@@ -20,7 +20,7 @@ plugins {
 
     // test
     id "com.adarshr.test-logger" version "4.0.0"
-    id "org.sonarqube" version "5.0.0.4638"
+    id "org.sonarqube" version "5.1.0.4882"
     id 'jacoco-report-aggregation'
 
     // helper
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 f89570a37..012226fb8 100644
--- a/core/src/main/java/io/kestra/core/runners/FlowableUtils.java
+++ b/core/src/main/java/io/kestra/core/runners/FlowableUtils.java
@@ -188,6 +188,10 @@ public class FlowableUtils {
             .toList();
     }
 
+    /**
+     * resolveParallelNexts will resolve both concurrent values and subtasks
+     * For only concurrent values, see resolveConcurrentNexts()
+     */
     public static List<NextTaskRun> resolveParallelNexts(
         Execution execution,
         List<ResolvedTask> tasks,
@@ -204,6 +208,74 @@ public class FlowableUtils {
         );
     }
 
+    /**
+     * resolveConcurrentNexts will resolve concurrent values
+     * For both concurrent vales and subtasks, see resolveParallelNexts()
+     */
+    public static List<NextTaskRun> resolveConcurrentNexts(
+        Execution execution,
+        List<ResolvedTask> tasks,
+        List<ResolvedTask> errors,
+        TaskRun parentTaskRun,
+        Integer concurrency
+    ) {
+        if (execution.getState().getCurrent() == State.Type.KILLING) {
+            return Collections.emptyList();
+        }
+
+        List<ResolvedTask> allTasks = execution.findTaskDependingFlowState(
+            tasks,
+            errors,
+            parentTaskRun
+        );
+
+        // all tasks run
+        List<TaskRun> taskRuns = execution.findTaskRunByTasks(allTasks, parentTaskRun);
+
+        // find all non-terminated
+        long nonTerminatedCount = taskRuns
+            .stream()
+            .filter(taskRun -> !taskRun.getState().isTerminated())
+            .count();
+
+        if (concurrency > 0 && nonTerminatedCount >= concurrency) {
+            return Collections.emptyList();
+        }
+
+        long concurrencySlots = concurrency - nonTerminatedCount;
+
+        // first one
+        if (taskRuns.isEmpty()) {
+            Map<String, List<ResolvedTask>> collect = allTasks
+                .stream()
+                .collect(Collectors.groupingBy(resolvedTask -> resolvedTask.getValue(), () -> new LinkedHashMap<>(), Collectors.toList()));
+            return collect.values().stream()
+                .limit(concurrencySlots)
+                .map(resolvedTasks -> resolvedTasks.getFirst().toNextTaskRun(execution))
+                .toList()
+                .reversed();
+        }
+
+        // start as many tasks as we have concurrency slots
+        Map<String, List<ResolvedTask>> collect = allTasks
+            .stream()
+            .collect(Collectors.groupingBy(resolvedTask -> resolvedTask.getValue(), () -> new LinkedHashMap<>(), Collectors.toList()));
+        return collect.values().stream()
+            .map(resolvedTasks -> filterCreated(resolvedTasks, taskRuns, parentTaskRun))
+            .filter(resolvedTasks -> !resolvedTasks.isEmpty())
+            .limit(concurrencySlots)
+            .map(resolvedTasks -> resolvedTasks.getFirst().toNextTaskRun(execution))
+            .toList();
+    }
+
+    private static List<ResolvedTask> filterCreated(List<ResolvedTask> tasks, List<TaskRun> taskRuns, TaskRun parentTaskRun) {
+        return tasks.stream()
+            .filter(resolvedTask -> taskRuns.stream()
+                .noneMatch(taskRun -> FlowableUtils.isTaskRunFor(resolvedTask, taskRun, parentTaskRun))
+            )
+            .toList();
+    }
+
     public static List<NextTaskRun> resolveDagNexts(
         Execution execution,
         List<ResolvedTask> tasks,
@@ -265,15 +337,6 @@ public class FlowableUtils {
         // all tasks run
         List<TaskRun> taskRuns = execution.findTaskRunByTasks(currentTasks, parentTaskRun);
 
-        // find all not created tasks
-        List<ResolvedTask> notFinds = currentTasks
-            .stream()
-            .filter(resolvedTask -> taskRuns
-                .stream()
-                .noneMatch(taskRun -> FlowableUtils.isTaskRunFor(resolvedTask, taskRun, parentTaskRun))
-            )
-            .toList();
-
         // find all running and deal concurrency
         long runningCount = taskRuns
             .stream()
@@ -284,6 +347,15 @@ public class FlowableUtils {
             return Collections.emptyList();
         }
 
+        // find all not created tasks
+        List<ResolvedTask> notFinds = currentTasks
+            .stream()
+            .filter(resolvedTask -> taskRuns
+                .stream()
+                .noneMatch(taskRun -> FlowableUtils.isTaskRunFor(resolvedTask, taskRun, parentTaskRun))
+            )
+            .toList();
+
         // first created, leave
         Optional<TaskRun> lastCreated = execution.findLastCreated(taskRuns);
 
diff --git a/core/src/main/java/io/kestra/plugin/core/flow/ForEach.java b/core/src/main/java/io/kestra/plugin/core/flow/ForEach.java
new file mode 100644
index 000000000..1ec76f34f
--- /dev/null
+++ b/core/src/main/java/io/kestra/plugin/core/flow/ForEach.java
@@ -0,0 +1,191 @@
+package io.kestra.plugin.core.flow;
+
+import io.kestra.core.exceptions.IllegalVariableEvaluationException;
+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.NextTaskRun;
+import io.kestra.core.models.executions.TaskRun;
+import io.kestra.core.models.flows.State;
+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.VoidOutput;
+import io.kestra.core.runners.FlowableUtils;
+import io.kestra.core.runners.RunContext;
+import io.kestra.core.utils.GraphUtils;
+import io.swagger.v3.oas.annotations.media.Schema;
+import jakarta.validation.constraints.NotNull;
+import lombok.*;
+import lombok.experimental.SuperBuilder;
+
+import java.util.List;
+import java.util.Optional;
+
+@SuperBuilder
+@ToString
+@EqualsAndHashCode
+@Getter
+@NoArgsConstructor
+@Schema(
+    title = "For each value in the list, execute one or more tasks sequentially.",
+    description = "The list of `tasks` will be executed for each value sequentially, but values can be executed concurrently if concurrency is set to more than one." +
+        "The values must be a valid JSON string representing an array, e.g. a list of strings `[\"value1\", \"value2\"]` or a list of dictionaries `[{\"key\": \"value1\"}, {\"key\": \"value2\"}]`. \n\n" +
+        "You can access the current iteration value using the variable `{{ taskrun.value }}`. " +
+        "The task list will be executed sequentially for each value, if you want them to be executed in parallel you can use a Parallel task.\n\n" +
+        "We highly recommend triggering a subflow for each value. " +
+        "This allows much better scalability and modularity. Check the [flow best practices documentation](https://kestra.io/docs/developer-guide/best-practices) " +
+        "and the [following Blueprint](https://kestra.io/blueprints/128-run-a-subflow-for-each-value-in-parallel-and-wait-for-their-completion-recommended-pattern-to-iterate-over-hundreds-or-thousands-of-list-items) " +
+        "for more details."
+)
+@Plugin(
+    examples = {
+        @Example(
+            full = true,
+            title = "The taskrun.value from the `each_sequential` task is available only to immediate child tasks such as the `before_if` and the `if` tasks. To access the taskrun value in child tasks of the `if` task (such as in the `after_if` task), you need to use the syntax `{{ parent.taskrun.value }}` as this allows you to access the taskrun value of the parent task `each_sequential`.",
+            code = """
+                id: loop_example
+                namespace: company.team
+
+                tasks:
+                  - id: for_each
+                    type: io.kestra.plugin.core.flow.ForEach
+                    values: ["value 1", "value 2", "value 3"]
+                    tasks:
+                      - id: before_if
+                        type: io.kestra.plugin.core.debug.Return
+                        format: 'Before if {{ taskrun.value }}'
+                      - id: if
+                        type: io.kestra.plugin.core.flow.If
+                        condition: '{{ taskrun.value == "value 2" }}'
+                        then:
+                          - id: after_if
+                            type: io.kestra.plugin.core.debug.Return
+                            format: 'After if {{ parent.taskrun.value }}'"""
+        ),
+        @Example(
+            full = true,
+            title = "This task shows that the value can be a bullet-style list. The task iterates over the list of values and executes the `each-value` child task for each value. It also process the values concurrently 2 by 2.",
+            code = {
+                "id: for_each",
+                "namespace: company.team",
+                "",
+                "tasks:",
+                "  - id: for_each",
+                "    type: io.kestra.plugin.core.flow.ForEach",
+                "    values: ",
+                "      - value 1",
+                "      - value 2",
+                "      - value 3",
+                "    concurrency: 2",
+                "    tasks:",
+                "      - id: each-value",
+                "        type: io.kestra.plugin.core.debug.Return",
+                "        format: \"{{ task.id }} with value '{{ taskrun.value }}'\"",
+            }
+        ),
+        @Example(
+            full = true,
+            title = "This task shows how to process the sub-tasks in parallel.",
+            code = """
+                id: loop_example
+                namespace: company.team
+
+                tasks:
+                  - id: for_each
+                    type: io.kestra.plugin.core.flow.ForEach
+                    values: ["value 1", "value 2", "value 3"]
+                    concurrency: 2
+                    tasks:
+                      - id: parallel
+                        type: io.kestra.plugin.core.flow.Parallel
+                        tasks:
+                        - id: log
+                          type: io.kestra.plugin.core.log.Log
+                          message: Processing
+                        - id: shell
+                          type: io.kestra.plugin.scripts.shell.Commands
+                          commands:
+                            - sleep 5"""
+        ),
+    }
+)
+public class ForEach extends Sequential implements FlowableTask<VoidOutput> {
+    @NotNull
+    @PluginProperty(dynamic = true)
+    @Schema(
+        title = "The list of values for this task.",
+        description = "The values car be passed as a string, a list of strings, or a list of objects.",
+        oneOf = {String.class, Object[].class}
+    )
+    private Object values;
+
+    @NotNull
+    @Builder.Default
+    @Schema(
+        title = "Number of concurrent values that can be running at any point in time.",
+        description = "If the concurrency is `0`, no limit exist and all the values will start at the same time."
+    )
+    @PluginProperty
+    private final Integer concurrency = 1;
+
+    @Override
+    public GraphCluster tasksTree(Execution execution, TaskRun taskRun, List<String> parentValues) throws IllegalVariableEvaluationException {
+        GraphCluster subGraph = new GraphCluster(this, taskRun, parentValues, RelationType.DYNAMIC);
+
+        GraphUtils.parallel(
+            subGraph,
+            this.getTasks(),
+            this.getErrors(),
+            taskRun,
+            execution
+        );
+
+        return subGraph;
+    }
+
+    @Override
+    public List<ResolvedTask> childTasks(RunContext runContext, TaskRun parentTaskRun) throws IllegalVariableEvaluationException {
+        return FlowableUtils.resolveEachTasks(runContext, parentTaskRun, this.getTasks(), this.values);
+    }
+
+    @Override
+    public Optional<State.Type> resolveState(RunContext runContext, Execution execution, TaskRun parentTaskRun) throws IllegalVariableEvaluationException {
+        List<ResolvedTask> childTasks = this.childTasks(runContext, parentTaskRun);
+
+        if (childTasks.isEmpty()) {
+            return Optional.of(State.Type.SUCCESS);
+        }
+
+        return FlowableUtils.resolveState(
+            execution,
+            childTasks,
+            FlowableUtils.resolveTasks(this.getErrors(), parentTaskRun),
+            parentTaskRun,
+            runContext,
+            this.isAllowFailure()
+        );
+    }
+
+    @Override
+    public List<NextTaskRun> resolveNexts(RunContext runContext, Execution execution, TaskRun parentTaskRun) throws IllegalVariableEvaluationException {
+        if (this.concurrency == 1) {
+            return FlowableUtils.resolveSequentialNexts(
+                execution,
+                this.childTasks(runContext, parentTaskRun),
+                FlowableUtils.resolveTasks(this.errors, parentTaskRun),
+                parentTaskRun
+            );
+        }
+
+        return FlowableUtils.resolveConcurrentNexts(
+            execution,
+            FlowableUtils.resolveEachTasks(runContext, parentTaskRun, this.getTasks(), this.values),
+            FlowableUtils.resolveTasks(this.errors, parentTaskRun),
+            parentTaskRun,
+            this.concurrency
+        );
+    }
+}
diff --git a/core/src/main/java/io/kestra/plugin/core/namespace/UploadFiles.java b/core/src/main/java/io/kestra/plugin/core/namespace/UploadFiles.java
index c7c9d5a96..386402393 100644
--- a/core/src/main/java/io/kestra/plugin/core/namespace/UploadFiles.java
+++ b/core/src/main/java/io/kestra/plugin/core/namespace/UploadFiles.java
@@ -1,11 +1,15 @@
 package io.kestra.plugin.core.namespace;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonMappingException;
+import io.kestra.core.exceptions.IllegalVariableEvaluationException;
 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.tasks.RunnableTask;
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.runners.RunContext;
+import io.kestra.core.serializers.JacksonMapper;
 import io.kestra.core.storages.Namespace;
 import io.kestra.core.utils.FileUtils;
 import io.swagger.v3.oas.annotations.media.Schema;
@@ -75,8 +79,9 @@ public class UploadFiles extends Task implements RunnableTask<UploadFiles.Output
     @Schema(
         title = "A list of files.",
         description = "This can be a list of strings, where each string can be either a regex glob pattern or a file path. Providing a list requires specifying a `destination` where files will be stored.\n" +
-                    "This can also be a map where you can provide a specific destination path for a URI, which can be useful if you need to rename a file or move it to a different folder.",
-        anyOf = {List.class, Map.class}
+                    "This can also be a map where you can provide a specific destination path for a URI, which can be useful if you need to rename a file or move it to a different folder.\n" +
+                    "Finally, this can also be a string if its a pebble expression that render to a list or a map, or a single URI (that will be handle like a list of one).",
+        anyOf = {List.class, Map.class, String.class}
     )
     @PluginProperty(dynamic = true)
     private Object files;
@@ -105,10 +110,12 @@ public class UploadFiles extends Task implements RunnableTask<UploadFiles.Output
 
         final Namespace storageNamespace = runContext.storage().namespace(renderedNamespace);
 
+        if (files instanceof String filesString) {
+            // check if rendered filesString is a map, a list or a string
+            files = renderFilesString(runContext, filesString);
+        }
+
         if (files instanceof List<?> filesList) {
-            if (renderedDestination == null) {
-                throw new RuntimeException("Destination must be set when providing a List for the `files` property.");
-            }
             filesList = runContext.render((List<String>) filesList);
 
             final List<String> regexs = new ArrayList<>();
@@ -150,6 +157,9 @@ public class UploadFiles extends Task implements RunnableTask<UploadFiles.Output
                     throw new IllegalArgumentException("files must be a List<String> or a Map<String, String>");
                 }
             }
+        } else {
+            throw new IllegalArgumentException("Files must be a List<String> or a Map<String, String>");
+
         }
 
         return Output.builder().build();
@@ -161,4 +171,17 @@ public class UploadFiles extends Task implements RunnableTask<UploadFiles.Output
         private final Map<String, URI> files;
     }
 
+    private Object renderFilesString(RunContext runContext, String filesString) throws IllegalVariableEvaluationException, JsonProcessingException {
+        String rendered = runContext.render(filesString);
+        if (runContext.storage().isFileExist(URI.create(rendered))) {
+            return List.of(rendered);
+        } else {
+            try {
+                return JacksonMapper.ofJson().readValue(rendered, Map.class);
+            } catch (JsonMappingException e) {
+                return JacksonMapper.ofJson().readValue(rendered, List.class);
+            }
+        }
+    }
+
 }
diff --git a/platform/build.gradle b/platform/build.gradle
index b7283ff83..183bebc7a 100644
--- a/platform/build.gradle
+++ b/platform/build.gradle
@@ -115,7 +115,7 @@ dependencies {
         api group: 'org.exparity', name: 'hamcrest-date', version: '2.0.8'
         api 'com.github.tomakehurst:wiremock-jre8:3.0.1'
         api "org.apache.kafka:kafka-streams-test-utils:$kafkaVersion"
-        api "com.microsoft.playwright:playwright:1.44.0"
+        api "com.microsoft.playwright:playwright:1.45.0"
 
         // Kestra components
         api "io.kestra:core:$version"


Test patch:
diff --git a/core/src/test/java/io/kestra/plugin/core/flow/ForEachTest.java b/core/src/test/java/io/kestra/plugin/core/flow/ForEachTest.java
new file mode 100644
index 000000000..83548e975
--- /dev/null
+++ b/core/src/test/java/io/kestra/plugin/core/flow/ForEachTest.java
@@ -0,0 +1,38 @@
+package io.kestra.plugin.core.flow;
+
+import io.kestra.core.models.executions.Execution;
+import io.kestra.core.models.flows.State;
+import io.kestra.core.runners.AbstractMemoryRunnerTest;
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.TimeoutException;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.hasSize;
+import static org.hamcrest.Matchers.is;
+
+class ForEachTest  extends AbstractMemoryRunnerTest {
+    @Test
+    void nonConcurrent() throws TimeoutException {
+        Execution execution = runnerUtils.runOne(null, "io.kestra.tests", "foreach-non-concurrent");
+
+        assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
+        assertThat(execution.getTaskRunList(), hasSize(7));
+    }
+
+    @Test
+    void concurrent() throws TimeoutException {
+        Execution execution = runnerUtils.runOne(null, "io.kestra.tests", "foreach-concurrent");
+
+        assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
+        assertThat(execution.getTaskRunList(), hasSize(7));
+    }
+
+    @Test
+    void concurrentWithParallel() throws TimeoutException {
+        Execution execution = runnerUtils.runOne(null, "io.kestra.tests", "foreach-concurrent-parallel");
+
+        assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
+        assertThat(execution.getTaskRunList(), hasSize(10));
+    }
+}
\ No newline at end of file
diff --git a/core/src/test/resources/flows/valids/foreach-concurrent-parallel.yaml b/core/src/test/resources/flows/valids/foreach-concurrent-parallel.yaml
new file mode 100644
index 000000000..9f0090139
--- /dev/null
+++ b/core/src/test/resources/flows/valids/foreach-concurrent-parallel.yaml
@@ -0,0 +1,18 @@
+id: foreach-concurrent-parallel
+namespace: io.kestra.tests
+
+tasks:
+  - id: for_each
+    type: io.kestra.plugin.core.flow.ForEach
+    values: ["value 1", "value 2", "value 3"]
+    concurrency: 2
+    tasks:
+      - id: parallel
+        type: io.kestra.plugin.core.flow.Parallel
+        tasks:
+          - id: log
+            type: io.kestra.plugin.core.log.Log
+            message: Processing
+          - id: shell
+            type: io.kestra.plugin.core.log.Log
+            message: 2nd task
\ No newline at end of file
diff --git a/core/src/test/resources/flows/valids/foreach-concurrent.yaml b/core/src/test/resources/flows/valids/foreach-concurrent.yaml
new file mode 100644
index 000000000..8ec32a51b
--- /dev/null
+++ b/core/src/test/resources/flows/valids/foreach-concurrent.yaml
@@ -0,0 +1,15 @@
+id: foreach-concurrent
+namespace: io.kestra.tests
+
+tasks:
+  - id: for_each
+    type: io.kestra.plugin.core.flow.ForEach
+    values: ["value 1", "value 2", "value 3"]
+    concurrency: 2
+    tasks:
+      - id: log
+        type: io.kestra.plugin.core.log.Log
+        message: Processing {{taskrun.value}}
+      - id: shell
+        type: io.kestra.plugin.core.log.Log
+        message: 2nd task
\ No newline at end of file
diff --git a/core/src/test/resources/flows/valids/foreach-non-concurrent.yaml b/core/src/test/resources/flows/valids/foreach-non-concurrent.yaml
new file mode 100644
index 000000000..ad6f9c92a
--- /dev/null
+++ b/core/src/test/resources/flows/valids/foreach-non-concurrent.yaml
@@ -0,0 +1,14 @@
+id: foreach-non-concurrent
+namespace: io.kestra.tests
+
+tasks:
+  - id: for_each
+    type: io.kestra.plugin.core.flow.ForEach
+    values: ["value 1", "value 2", "value 3"]
+    tasks:
+      - id: log
+        type: io.kestra.plugin.core.log.Log
+        message: Processing {{taskrun.value}}
+      - id: shell
+        type: io.kestra.plugin.core.log.Log
+        message: 2nd task
\ No newline at end of file
diff --git a/e2e-tests/build.gradle b/e2e-tests/build.gradle
index 5a23988c9..77dfcd9bf 100644
--- a/e2e-tests/build.gradle
+++ b/e2e-tests/build.gradle
@@ -4,7 +4,7 @@ configurations {
 
 dependencies {
     testImplementation project(':tests')
-    testImplementation("com.microsoft.playwright:playwright:1.44.0")
+    testImplementation("com.microsoft.playwright:playwright:1.45.0")
 }
 
 /**********************************************************************************************************************\