Fix/for each

core/src/main/java/io/kestra/core/runners/FlowableUtils.java

@@ -242,7 +242,7 @@ public static List<NextTaskRun> resolveConcurrentNexts(
             return Collections.emptyList();
         }
 
-        long concurrencySlots = concurrency - nonTerminatedCount;
+        long concurrencySlots = concurrency == 0 ? Integer.MAX_VALUE : concurrency - nonTerminatedCount;
 
         // first one
         if (taskRuns.isEmpty()) {

core/src/main/java/io/kestra/plugin/core/flow/ForEach.java

@@ -18,6 +18,7 @@
 import io.kestra.core.utils.GraphUtils;
 import io.swagger.v3.oas.annotations.media.Schema;
 import jakarta.validation.constraints.NotNull;
+import jakarta.validation.constraints.PositiveOrZero;
 import lombok.*;
 import lombok.experimental.SuperBuilder;
 
@@ -147,6 +148,7 @@ public class ForEach extends Sequential implements FlowableTask<VoidOutput> {
     )
     private Object values;
 
+    @PositiveOrZero
     @NotNull
     @Builder.Default
     @Schema(
@@ -164,13 +166,23 @@ public class ForEach extends Sequential implements FlowableTask<VoidOutput> {
     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
-        );
+        if (concurrencyLimit == 1) {
+            GraphUtils.sequential(
+                subGraph,
+                this.getTasks(),
+                this.getErrors(),
+                taskRun,
+                execution
+            );
+        } else {
+            GraphUtils.parallel(
+                subGraph,
+                this.getTasks(),
+                this.getErrors(),
+                taskRun,
+                execution
+            );
+        }
 
         return subGraph;
     }

core/src/test/java/io/kestra/plugin/core/flow/ForEachTest.java

@@ -35,4 +35,12 @@ void concurrentWithParallel() throws TimeoutException {
         assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
         assertThat(execution.getTaskRunList(), hasSize(10));
     }
+
+    @Test
+    void concurrentNoLimit() throws TimeoutException {
+        Execution execution = runnerUtils.runOne(null, "io.kestra.tests", "foreach-concurrent-no-limit");
+
+        assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
+        assertThat(execution.getTaskRunList(), hasSize(7));
+    }
 }

core/src/test/resources/flows/valids/foreach-concurrent-no-limit.yaml

@@ -0,0 +1,15 @@
+id: foreach-concurrent-no-limit
+namespace: io.kestra.tests
+
+tasks:
+  - id: for_each
+    type: io.kestra.plugin.core.flow.ForEach
+    values: ["value 1", "value 2", "value 3"]
+    concurrencyLimit: 0
+    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