fix(core): Continue WaitFor loop if tasks are not failed (#6572)

close #6031

core/src/main/java/io/kestra/core/models/flows/State.java

@@ -135,6 +135,11 @@ public boolean isTerminated() {
         return this.current.isTerminated();
     }
 
+    @JsonIgnore
+    public boolean isTerminatedNoFail() {
+        return this.current.isTerminatedNoFail();
+    }
+
     @JsonIgnore
     public boolean isRunning() {
         return this.current.isRunning();
@@ -216,6 +221,10 @@ public boolean isTerminated() {
             return this == Type.FAILED || this == Type.WARNING || this == Type.SUCCESS || this == Type.KILLED || this == Type.CANCELLED || this == Type.RETRIED || this == Type.SKIPPED;
         }
 
+        public boolean isTerminatedNoFail() {
+            return this == Type.WARNING || this == Type.SUCCESS || this == Type.RETRIED || this == Type.SKIPPED;
+        }
+
         public boolean isCreated() {
             return this == Type.CREATED || this == Type.RESTARTED;
         }

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

@@ -209,7 +209,7 @@ public boolean childTaskRunExecuted(Execution execution, TaskRun parentTaskRun)
             .stream()
             .filter(t -> t.getParentTaskRunId() != null
                 && t.getParentTaskRunId().equals(parentTaskRun.getId())
-                && t.getState().isSuccess()
+                && t.getState().isTerminatedNoFail()
             ).count() == tasks.size();
 
     }

core/src/test/java/io/kestra/core/runners/AbstractRunnerTest.java

@@ -76,7 +76,7 @@ public abstract class AbstractRunnerTest {
     protected ForEachItemCaseTest forEachItemCaseTest;
 
     @Inject
-    private WaitForCaseTest waitForTestCaseTest;
+    protected WaitForCaseTest waitForTestCaseTest;
 
     @Inject
     private FlowConcurrencyCaseTest flowConcurrencyCaseTest;

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

@@ -61,4 +61,11 @@ public void waitforMultipleTasksFailed() throws TimeoutException, QueueException
         assertThat(execution.getState().getCurrent(), is(State.Type.FAILED));
         assertThat(execution.getTaskRunList().getLast().attemptNumber(), is(1));
     }
+
+    public void waitForChildTaskWarning() throws TimeoutException, QueueException {
+        Execution execution = runnerUtils.runOne(null, "io.kestra.tests", "waitfor-child-task-warning");
+
+        assertThat(execution.getState().getCurrent(), is(State.Type.FAILED));
+        assertThat((Integer) execution.getTaskRunList().getFirst().getOutputs().get("iterationCount"), greaterThan(1));
+    }
 }

core/src/test/resources/flows/valids/waitfor-child-task-warning.yaml

@@ -0,0 +1,22 @@
+id: waitfor-child-task-warning
+namespace: io.kestra.tests
+
+tasks:
+  - id: loop
+    type: io.kestra.plugin.core.flow.WaitFor
+    condition: "{{ outputs.check_migration_task.values['test'] == 'FINISHED' }}"
+    failOnMaxReached: true
+    checkFrequency:
+      interval: PT5S
+      maxDuration: PT10S
+    tasks:
+
+      ## forcing a Warning
+      - id: allow_failure
+        allowFailure: true
+        type: io.kestra.plugin.core.execution.Fail
+
+      - id: check_migration_task
+        type: io.kestra.plugin.core.output.OutputValues
+        values:
+          test: "ok"

jdbc/src/test/java/io/kestra/jdbc/runner/JdbcRunnerTest.java

@@ -37,6 +37,12 @@ public abstract class JdbcRunnerTest extends AbstractRunnerTest {
     @Inject
     private JdbcTestUtils jdbcTestUtils;
 
+    @Test
+    @LoadFlows({"flows/valids/waitfor-child-task-warning.yaml"})
+    void waitForChildTaskWarning() throws Exception {
+        waitForTestCaseTest.waitForChildTaskWarning();
+    }
+
     @Test
     @LoadFlows({"flows/valids/inputs-large.yaml"})
     void flowTooLarge() throws Exception {