Repository: kestra-io/kestra (java)

Issue:
Add a runnable task `WaitFor` that will run a task until a specific condition is true
### Feature description

Syntax:

```yaml
id: check_for_job_status
namespace: dev

tasks:
  - id: databricks_job
    type: io.kestra.core.tasks.flows.WaitFor
    condition: "{{ (outputs.poll.body | jq('.state.result_state') | first) == 'SUCCESS' }}"
    task:
      id: poll
      type: io.kestra.plugin.fs.http.Request
      uri: https://reqres.in/api/job-status
      method: GET
      contentType: application/json
    checkFrequency:
      interval: PT1S # type: Duration
      maxDuration: PT1H # type: Duration
      maxAttempts: 50 # type: int
```

Icon:
![image (1)](https://github.com/kestra-io/kestra/assets/86264395/17af6c40-39ba-49e0-85f8-2ebf2ddd2065)


The reason for this task (and the implementation):
1. Implementing a `WhileLoop` that would satisfy multiple use cases in one go proved challenging as we cannot promise this level of statefulness 
2. Implementing a flowable task that could execute multiple tasks within the loop could break the executor 
3. We don't implement all the same functionality as we do on the retry in the `checkFrequency` because this task is meant for waiting until the condition is satisfied (blocking calls) rather than for fetching data from APIs, which might require exponential retries.

For those reasons, we decided to implement this as a runnable task that can execute a single task. Later, we may explore the `WhileLoop` once we figure out a solution to the second issue as part of the `ForEachItem` redesign.

Pull request:
feat(core): new WaitFor task
closes #3587

To do : 
- Find a way not to require id (Custom serializer ? May conflict with schemas)

Example:
```yaml
id: waitfor
namespace: io.kestra.tests

tasks:
  - id: waitfor
    type: io.kestra.core.tasks.flows.WaitFor
    condition: "{{ outputs.output_values.values.count != '4'}}"
    maxIterations: 5
    task:
      id: output_values
      type: io.kestra.core.tasks.outputs.OutputValues
      values:
        count: "{{ outputs.waitfor.iterationCount + 1 }}"
```

Patch:
diff --git a/core/src/main/java/io/kestra/core/models/flows/State.java b/core/src/main/java/io/kestra/core/models/flows/State.java
index 17a497017..28b99a51c 100644
--- a/core/src/main/java/io/kestra/core/models/flows/State.java
+++ b/core/src/main/java/io/kestra/core/models/flows/State.java
@@ -154,6 +154,11 @@ public class State {
         return this.current.isRetrying();
     }
 
+    @JsonIgnore
+    public boolean isSuccess() {
+        return this.current.isSuccess();
+    }
+
     @JsonIgnore
     public boolean isRestartable() {
         return this.current.isFailed() || this.isPaused();
@@ -205,6 +210,10 @@ public class State {
             return this == Type.RETRYING || this == Type.RETRIED;
         }
 
+        public boolean isSuccess() {
+            return this == Type.SUCCESS;
+        }
+
     }
 
     @Value
diff --git a/core/src/main/java/io/kestra/core/models/tasks/FlowableTask.java b/core/src/main/java/io/kestra/core/models/tasks/FlowableTask.java
index a1f489b60..29c4a9f3d 100644
--- a/core/src/main/java/io/kestra/core/models/tasks/FlowableTask.java
+++ b/core/src/main/java/io/kestra/core/models/tasks/FlowableTask.java
@@ -1,16 +1,15 @@
 package io.kestra.core.models.tasks;
 
-import io.kestra.core.models.annotations.PluginProperty;
-import io.kestra.core.models.hierarchies.AbstractGraph;
-import io.swagger.v3.oas.annotations.media.Schema;
 import io.kestra.core.exceptions.IllegalVariableEvaluationException;
+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.AbstractGraph;
 import io.kestra.core.runners.FlowableUtils;
 import io.kestra.core.runners.RunContext;
+import io.swagger.v3.oas.annotations.media.Schema;
 
 import java.util.List;
 import java.util.Optional;
diff --git a/core/src/main/java/io/kestra/core/runners/ExecutionDelay.java b/core/src/main/java/io/kestra/core/runners/ExecutionDelay.java
index c56702333..bae0e0296 100644
--- a/core/src/main/java/io/kestra/core/runners/ExecutionDelay.java
+++ b/core/src/main/java/io/kestra/core/runners/ExecutionDelay.java
@@ -42,6 +42,7 @@ public class ExecutionDelay {
     public enum DelayType {
         RESUME_FLOW,
         RESTART_FAILED_TASK,
-        RESTART_FAILED_FLOW
+        RESTART_FAILED_FLOW,
+        CONTINUE_FLOWABLE
     }
 }
diff --git a/core/src/main/java/io/kestra/core/runners/ExecutorService.java b/core/src/main/java/io/kestra/core/runners/ExecutorService.java
index 13c54de67..eacd9a46c 100644
--- a/core/src/main/java/io/kestra/core/runners/ExecutorService.java
+++ b/core/src/main/java/io/kestra/core/runners/ExecutorService.java
@@ -11,6 +11,7 @@ import io.kestra.core.models.tasks.retrys.AbstractRetry;
 import io.kestra.core.services.ConditionService;
 import io.kestra.core.services.ExecutionService;
 import io.kestra.core.services.LogService;
+import io.kestra.core.tasks.flows.WaitFor;
 import io.kestra.core.tasks.flows.Pause;
 import io.kestra.core.tasks.flows.WorkingDirectory;
 import io.micronaut.context.ApplicationContext;
@@ -523,7 +524,7 @@ public class ExecutorService {
                             ExecutionDelay.DelayType.RESTART_FAILED_TASK);
                     executionDelays.add(executionDelayBuilder.build());
                     executor.withExecution(behavior.equals(AbstractRetry.Behavior.CREATE_NEW_EXECUTION) ?
-                            executionService.markWithTaskRunAs(executor.getExecution(), taskRun.getId(), State.Type.RETRIED, true):
+                            executionService.markWithTaskRunAs(executor.getExecution(), taskRun.getId(), State.Type.RETRIED, true) :
                             executionService.markWithTaskRunAs(executor.getExecution(), taskRun.getId(), State.Type.RETRYING, false),
                         "handleRetryTask");
                     // Prevent workerTaskResult of flowable to be sent
@@ -533,6 +534,30 @@ public class ExecutorService {
                     }
                 }
             }
+            // WaitFor case
+            else if (task instanceof WaitFor waitFor && taskRun.getState().isRunning()) {
+                TaskRun childTaskRun = waitFor.getChildTaskRun(executor.getExecution(), taskRun).orElse(null);
+                if (childTaskRun != null && childTaskRun.getState().isSuccess()) {
+                    Output newOutput = waitFor.outputs(taskRun);
+                    TaskRun updatedTaskRun = taskRun.withOutputs(newOutput.toMap());
+                    RunContext runContext = runContextFactory.of(executor.getFlow(), task, executor.getExecution(), updatedTaskRun);
+                    List<NextTaskRun> next = ((FlowableTask<?>) task).resolveNexts(runContext, executor.getExecution(), updatedTaskRun);
+                    Instant nextDate = waitFor.nextExecutionDate(runContext, executor.getExecution(), updatedTaskRun);
+                    if (next.isEmpty()) {
+                        return executor;
+                    } else if (nextDate != null && !next.isEmpty()) {
+                        executionDelays.add(ExecutionDelay.builder()
+                            .taskRunId(taskRun.getId())
+                            .executionId(executor.getExecution().getId())
+                            .date(nextDate)
+                            .state(State.Type.RUNNING)
+                            .delayType(ExecutionDelay.DelayType.CONTINUE_FLOWABLE)
+                            .build());
+                        Execution execution = executionService.pauseFlowable(executor.getExecution(), taskRun.getId(), newOutput);
+                        executor.withExecution(execution, "pauseLoop");
+                    }
+                }
+            }
 
             // If the task is retrying
             // make sure that the workerTaskResult of the parent task is not sent
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 a6cc6a64e..c41b78f99 100644
--- a/core/src/main/java/io/kestra/core/runners/FlowableUtils.java
+++ b/core/src/main/java/io/kestra/core/runners/FlowableUtils.java
@@ -88,6 +88,48 @@ public class FlowableUtils {
         return Collections.emptyList();
     }
 
+    public static List<NextTaskRun> resolveWaitForNext(
+        Execution execution,
+        List<ResolvedTask> tasks,
+        List<ResolvedTask> errors,
+        TaskRun parentTaskRun
+    ) {
+        List<ResolvedTask> currentTasks = execution.findTaskDependingFlowState(tasks, errors, parentTaskRun);
+
+        // nothing
+        if (currentTasks == null || currentTasks.isEmpty() || execution.getState().getCurrent() == State.Type.KILLING) {
+            return Collections.emptyList();
+        }
+
+        // first one
+        List<TaskRun> taskRuns = execution.findTaskRunByTasks(currentTasks, parentTaskRun);
+        if (taskRuns.isEmpty()) {
+            return Collections.singletonList(
+                currentTasks.getFirst().toNextTaskRun(execution)
+            );
+        }
+
+        // first created, leave
+        Optional<TaskRun> lastCreated = execution.findLastCreated(taskRuns);
+        if (lastCreated.isPresent()) {
+            return Collections.emptyList();
+        }
+
+        // have running, leave
+        Optional<TaskRun> lastRunning = execution.findLastRunning(taskRuns);
+        if (lastRunning.isPresent()) {
+            return Collections.emptyList();
+        }
+
+        // last success, find next
+        Optional<TaskRun> lastTerminated = execution.findLastTerminated(taskRuns);
+        if (lastTerminated.isPresent()) {
+            return Collections.singletonList(currentTasks.getFirst().toNextTaskRun(execution));
+        }
+
+        return Collections.emptyList();
+    }
+
     public static Optional<State.Type> resolveState(
         Execution execution,
         List<ResolvedTask> tasks,
diff --git a/core/src/main/java/io/kestra/core/services/ExecutionService.java b/core/src/main/java/io/kestra/core/services/ExecutionService.java
index b1a90797d..401cd0524 100644
--- a/core/src/main/java/io/kestra/core/services/ExecutionService.java
+++ b/core/src/main/java/io/kestra/core/services/ExecutionService.java
@@ -8,6 +8,7 @@ import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.flows.State;
 import io.kestra.core.models.hierarchies.AbstractGraphTask;
 import io.kestra.core.models.hierarchies.GraphCluster;
+import io.kestra.core.models.tasks.Output;
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.models.tasks.retrys.AbstractRetry;
 import io.kestra.core.queues.QueueFactoryInterface;
@@ -98,6 +99,48 @@ public class ExecutionService {
         return execution.withTaskRunList(newTaskRuns).withState(State.Type.RUNNING);
     }
 
+    public Execution retryFlowable(Execution execution, String flowableTaskRunId) {
+        List<TaskRun> newTaskRuns = execution
+            .getTaskRunList()
+            .stream()
+            .map(taskRun -> {
+                if (taskRun.getId().equals(flowableTaskRunId)) {
+                    return taskRun
+                        .withState(State.Type.RUNNING);
+                }
+
+                if (flowableTaskRunId.equals(taskRun.getParentTaskRunId())) {
+                    return taskRun.withState(State.Type.CREATED);
+                }
+
+                return taskRun;
+            })
+            .toList();
+
+        return execution.withTaskRunList(newTaskRuns).withState(State.Type.RUNNING);
+    }
+
+    public Execution pauseFlowable(Execution execution, String flowableTaskRunId, Output newOutput) {
+        List<TaskRun> newTaskRuns = execution
+            .getTaskRunList()
+            .stream()
+            .map(taskRun -> {
+                if (taskRun.getId().equals(flowableTaskRunId)) {
+                    var taskrunUpdated = taskRun
+                        .withState(State.Type.PAUSED);
+                    if (newOutput != null) {
+                        taskrunUpdated = taskrunUpdated.withOutputs(newOutput.toMap());
+                    }
+                    return taskrunUpdated;
+                }
+
+                return taskRun;
+            })
+            .toList();
+
+        return execution.withTaskRunList(newTaskRuns).withState(State.Type.PAUSED);
+    }
+
     public Execution restart(final Execution execution, @Nullable Integer revision) throws Exception {
         if (!(execution.getState().isTerminated() || execution.getState().isPaused())) {
             throw new IllegalStateException("Execution must be terminated to be restarted, " +
diff --git a/core/src/main/java/io/kestra/core/tasks/flows/WaitFor.java b/core/src/main/java/io/kestra/core/tasks/flows/WaitFor.java
new file mode 100644
index 000000000..e95563fd9
--- /dev/null
+++ b/core/src/main/java/io/kestra/core/tasks/flows/WaitFor.java
@@ -0,0 +1,241 @@
+package io.kestra.core.tasks.flows;
+
+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.AbstractGraph;
+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.runners.FlowableUtils;
+import io.kestra.core.runners.RunContext;
+import io.kestra.core.utils.GraphUtils;
+import io.kestra.core.utils.TruthUtils;
+import io.swagger.v3.oas.annotations.media.Schema;
+import jakarta.validation.Valid;
+import jakarta.validation.constraints.NotNull;
+import lombok.*;
+import lombok.experimental.SuperBuilder;
+import org.slf4j.Logger;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+@SuperBuilder
+@ToString
+@EqualsAndHashCode
+@Getter
+@NoArgsConstructor
+@Schema(
+    title = "Run a specific task repeatedly until the expected condition is met.",
+    description = """
+        Use this task if your downstream processing requires waiting for a specific HTTP response or a job to finish. 
+        You can access the task output in the `condition`.
+        
+        The `condition` is always checked after the task execution.
+        """
+)
+@Plugin(
+    examples = {
+        @Example(
+            full = true,
+            title = "Wait for a task to return a specific output",
+            code = """
+                id: exampleFlow
+                namespace: myteam
+                                
+                tasks:
+                  - id: waitFor
+                    type: io.kestra.core.tasks.flows.WaitFor
+                    condition: "{{ outputs.return.value != '4' }}"
+                    task:
+                      id: return
+                      type: io.kestra.core.tasks.debugs.Return
+                      format: "{{ outputs.waitfor.iterationCount }}"
+                """
+        )
+
+    }
+)
+public class WaitFor extends Task implements FlowableTask<WaitFor.Output> {
+    @Valid
+    protected List<Task> errors;
+
+    @Valid
+    @PluginProperty
+    @NotNull
+    private Task task;
+
+    @NotNull
+    @PluginProperty(dynamic = true)
+    @Schema(
+        title = "The condition to execute again the task that must be a boolean.",
+        description = "Boolean coercion allows 0, -0, null and '' to evaluate to false, all other values will evaluate to true."
+    )
+    private String condition;
+
+    @Schema(
+        title = "Maximum count of iterations."
+    )
+    @Builder.Default
+    private Integer maxIterations = 100;
+
+    @Schema(
+        title = "Maximum duration of the task."
+    )
+    @Builder.Default
+    private Duration maxDuration = Duration.ofHours(1);
+
+    @Schema(
+        title = "Interval between each iteration."
+    )
+    @Builder.Default
+    private Duration interval = Duration.ofSeconds(1);
+
+    @Schema(
+        title = "If true, the task will fail if the `maxIterations` or `maxDuration` are reached."
+    )
+    @Builder.Default
+    private Boolean failOnMaxReached = false;
+
+    @Override
+    public AbstractGraph tasksTree(Execution execution, TaskRun taskRun, List<String> parentValues) throws IllegalVariableEvaluationException {
+        GraphCluster subGraph = new GraphCluster(this, taskRun, parentValues, RelationType.SEQUENTIAL);
+
+        GraphUtils.sequential(
+            subGraph,
+            List.of(task),
+            this.errors,
+            taskRun,
+            execution
+        );
+
+        return subGraph;
+    }
+
+    @Override
+    public List<Task> allChildTasks() {
+        return Stream
+            .concat(
+                Stream.of(task),
+                this.getErrors() != null ? this.getErrors().stream() : Stream.empty()
+            )
+            .collect(Collectors.toList());
+    }
+
+    @Override
+    public List<ResolvedTask> childTasks(RunContext runContext, TaskRun parentTaskRun) throws IllegalVariableEvaluationException {
+        return FlowableUtils.resolveTasks(List.of(task), parentTaskRun);
+    }
+
+    @Override
+    public List<NextTaskRun> resolveNexts(RunContext runContext, Execution execution, TaskRun parentTaskRun) throws IllegalVariableEvaluationException {
+
+        return FlowableUtils.resolveWaitForNext(
+            execution,
+            this.childTasks(runContext, parentTaskRun),
+            FlowableUtils.resolveTasks(this.getErrors(), parentTaskRun),
+            parentTaskRun
+        );
+    }
+
+    public Instant nextExecutionDate(RunContext runContext, Execution execution, TaskRun parentTaskRun) throws IllegalVariableEvaluationException {
+        if (!this.reachedMaximums(runContext, execution, parentTaskRun, false)) {
+            String continueLoop = runContext.render(this.condition);
+            if (TruthUtils.isTruthy(continueLoop)) {
+
+                return Instant.now().plus(this.interval);
+            }
+        }
+
+        return null;
+    }
+
+    private boolean reachedMaximums(RunContext runContext, Execution execution, TaskRun parentTaskRun, Boolean printLog) {
+        TaskRun childTaskRun = this.getChildTaskRun(execution, parentTaskRun).orElse(null);
+        Logger logger = runContext.logger();
+
+        if (childTaskRun == null) {
+            return false;
+        }
+
+        if (this.maxIterations != null && childTaskRun.attemptNumber() >= this.maxIterations) {
+            if (printLog) {logger.warn("Max iterations reached");}
+            return true;
+        }
+
+        if (this.maxDuration != null &&
+            childTaskRun.getAttempts().getFirst().getState().getStartDate().plus(this.maxDuration).isBefore(Instant.now())) {
+            if (printLog) {logger.warn("Max duration reached");}
+
+            return true;
+        }
+
+        return false;
+    }
+
+    @Override
+    public Optional<State.Type> resolveState(RunContext runContext, Execution execution, TaskRun parentTaskRun) throws IllegalVariableEvaluationException {
+        TaskRun childTaskRun = this.getChildTaskRun(execution, parentTaskRun).orElse(null);
+        if (childTaskRun != null && nextExecutionDate(runContext, execution, parentTaskRun) != null) {
+            return Optional.of(State.Type.RUNNING);
+        }
+
+        if (childTaskRun != null && this.reachedMaximums(runContext, execution, parentTaskRun, true) && this.failOnMaxReached) {
+            return Optional.of(State.Type.FAILED);
+        }
+
+        return FlowableUtils.resolveState(
+            execution,
+            this.childTasks(runContext, parentTaskRun),
+            FlowableUtils.resolveTasks(this.getErrors(), parentTaskRun),
+            parentTaskRun,
+            runContext,
+            isAllowFailure()
+        );
+    }
+
+    public Optional<TaskRun> getChildTaskRun(Execution execution, TaskRun parentTaskRun) {
+        if (execution.getTaskRunList() == null) {
+            return Optional.empty();
+        }
+        return execution
+            .getTaskRunList()
+            .stream()
+            .filter(t -> t.getParentTaskRunId() != null && t.getParentTaskRunId().equals(parentTaskRun.getId()) && t.getState().isSuccess())
+            .findFirst();
+    }
+
+    @Override
+    public WaitFor.Output outputs(RunContext runContext) throws IllegalVariableEvaluationException {
+        return WaitFor.Output.builder()
+            .iterationCount(0)
+            .build();
+    }
+
+    public WaitFor.Output outputs(TaskRun parentTaskRun) throws IllegalVariableEvaluationException {
+        String value = parentTaskRun != null ?
+            parentTaskRun.getOutputs().get("iterationCount").toString() : "0";
+
+        return Output.builder()
+            .iterationCount(Integer.parseInt(value) + 1)
+            .build();
+    }
+
+    @Builder
+    @Getter
+    public static class Output implements io.kestra.core.models.tasks.Output {
+        private Integer iterationCount;
+    }
+}
diff --git a/core/src/main/resources/icons/io.kestra.core.tasks.flows.WaitFor.svg b/core/src/main/resources/icons/io.kestra.core.tasks.flows.WaitFor.svg
new file mode 100644
index 000000000..d1dec2f4c
--- /dev/null
+++ b/core/src/main/resources/icons/io.kestra.core.tasks.flows.WaitFor.svg
@@ -0,0 +1,1 @@
+<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 24 24"><path d="M11 8H13V14H11V8M12 20C8.13 20 5 16.87 5 13S8.13 6 12 6C15.19 6 17.88 8.14 18.72 11.05C19 11 19.24 11 19.5 11C19.94 11 20.38 11.05 20.8 11.13C20.5 9.74 19.89 8.46 19.03 7.39L20.45 5.97C20 5.46 19.55 5 19.04 4.56L17.62 6C16.07 4.74 14.12 4 12 4C7.03 4 3 8.03 3 13S7.03 22 12 22C12.87 22 13.7 21.87 14.5 21.65C14.05 21.11 13.7 20.5 13.45 19.85C13 19.95 12.5 20 12 20M15 1H9V3H15V1M19 13.5V12L16.75 14.25L19 16.5V15C20.38 15 21.5 16.12 21.5 17.5C21.5 17.9 21.41 18.28 21.24 18.62L22.33 19.71C22.75 19.08 23 18.32 23 17.5C23 15.29 21.21 13.5 19 13.5M19 20C17.62 20 16.5 18.88 16.5 17.5C16.5 17.1 16.59 16.72 16.76 16.38L15.67 15.29C15.25 15.92 15 16.68 15 17.5C15 19.71 16.79 21.5 19 21.5V23L21.25 20.75L19 18.5V20Z" /></svg>
\ No newline at end of file
diff --git a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java
index 5dccf3f06..c0dbc87a2 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java
@@ -862,6 +862,10 @@ public class JdbcExecutor implements ExecutorInterface, Service {
                         Execution newExecution = executionService.replay(executor.getExecution(), null, null);
                         executor = executor.withExecution(newExecution, "retryFailedFlow");
                     }
+                    else if (executionDelay.getDelayType().equals(ExecutionDelay.DelayType.CONTINUE_FLOWABLE)) {
+                        Execution execution  = executionService.retryFlowable(executor.getExecution(), executionDelay.getTaskRunId());
+                        executor = executor.withExecution(execution, "continueLoop");
+                    }
                 } catch (Exception e) {
                     executor = handleFailedExecutionFromExecutor(executor, e);
                 }


Test patch:
diff --git a/core/src/test/java/io/kestra/core/tasks/flows/WaitForCaseTest.java b/core/src/test/java/io/kestra/core/tasks/flows/WaitForCaseTest.java
new file mode 100644
index 000000000..3c96f9a72
--- /dev/null
+++ b/core/src/test/java/io/kestra/core/tasks/flows/WaitForCaseTest.java
@@ -0,0 +1,45 @@
+package io.kestra.core.tasks.flows;
+
+import io.kestra.core.models.executions.Execution;
+import io.kestra.core.models.flows.State;
+import io.kestra.core.repositories.FlowRepositoryInterface;
+import io.kestra.core.runners.RunnerUtils;
+import jakarta.inject.Inject;
+
+import java.util.concurrent.TimeoutException;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.*;
+
+public class WaitForCaseTest {
+    @Inject
+    FlowRepositoryInterface flowRepository;
+
+    @Inject
+    protected RunnerUtils runnerUtils;
+
+    public void waitfor() throws TimeoutException {
+        Execution execution = runnerUtils.runOne(null, "io.kestra.tests", "waitfor");
+
+        assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
+        assertThat(execution.getTaskRunList().getFirst().getOutputs(), notNullValue());
+    }
+
+    public void waitforMaxIterations() throws TimeoutException {
+        Execution execution = runnerUtils.runOne(null, "io.kestra.tests", "waitfor-max-iterations");
+
+        assertThat(execution.getState().getCurrent(), is(State.Type.FAILED));
+    }
+
+    public void waitforMaxDuration() throws TimeoutException {
+        Execution execution = runnerUtils.runOne(null, "io.kestra.tests", "waitfor-max-duration");
+
+        assertThat(execution.getState().getCurrent(), is(State.Type.FAILED));
+    }
+
+    public void waitforNoSuccess() throws TimeoutException {
+        Execution execution = runnerUtils.runOne(null, "io.kestra.tests", "waitfor-no-success");
+
+        assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
+    }
+}
diff --git a/core/src/test/resources/flows/valids/waitfor-max-duration.yaml b/core/src/test/resources/flows/valids/waitfor-max-duration.yaml
new file mode 100644
index 000000000..d9325f52c
--- /dev/null
+++ b/core/src/test/resources/flows/valids/waitfor-max-duration.yaml
@@ -0,0 +1,14 @@
+id: waitfor-max-duration
+namespace: io.kestra.tests
+
+tasks:
+  - id: waitfor
+    type: io.kestra.core.tasks.flows.WaitFor
+    condition: "{{ outputs.output_values.values.count != '-5'}}"
+    maxDuration: PT5S
+    failOnMaxReached: true
+    task:
+      id: output_values
+      type: io.kestra.core.tasks.outputs.OutputValues
+      values:
+        count: "{{ outputs.waitfor.iterationCount + 1 }}"
\ No newline at end of file
diff --git a/core/src/test/resources/flows/valids/waitfor-max-iterations.yaml b/core/src/test/resources/flows/valids/waitfor-max-iterations.yaml
new file mode 100644
index 000000000..b0cb69c66
--- /dev/null
+++ b/core/src/test/resources/flows/valids/waitfor-max-iterations.yaml
@@ -0,0 +1,14 @@
+id: waitfor-max-iterations
+namespace: io.kestra.tests
+
+tasks:
+  - id: waitfor
+    type: io.kestra.core.tasks.flows.WaitFor
+    condition: "{{ outputs.output_values.values.count != '10'}}"
+    maxIterations: 3
+    failOnMaxReached: true
+    task:
+      id: output_values
+      type: io.kestra.core.tasks.outputs.OutputValues
+      values:
+        count: "{{ outputs.waitfor.iterationCount + 1 }}"
\ No newline at end of file
diff --git a/core/src/test/resources/flows/valids/waitfor-no-success.yaml b/core/src/test/resources/flows/valids/waitfor-no-success.yaml
new file mode 100644
index 000000000..00084a1ce
--- /dev/null
+++ b/core/src/test/resources/flows/valids/waitfor-no-success.yaml
@@ -0,0 +1,13 @@
+id: waitfor-no-success
+namespace: io.kestra.tests
+
+tasks:
+  - id: waitfor
+    type: io.kestra.core.tasks.flows.WaitFor
+    condition: "{{ outputs.output_values.values.count != '6'}}"
+    maxIterations: 5
+    task:
+      id: output_values
+      type: io.kestra.core.tasks.outputs.OutputValues
+      values:
+        count: "{{ outputs.waitfor.iterationCount + 1 }}"
\ No newline at end of file
diff --git a/core/src/test/resources/flows/valids/waitfor.yaml b/core/src/test/resources/flows/valids/waitfor.yaml
new file mode 100644
index 000000000..a782f7da2
--- /dev/null
+++ b/core/src/test/resources/flows/valids/waitfor.yaml
@@ -0,0 +1,13 @@
+id: waitfor
+namespace: io.kestra.tests
+
+tasks:
+  - id: waitfor
+    type: io.kestra.core.tasks.flows.WaitFor
+    condition: "{{ outputs.output_values.values.count != '4'}}"
+    maxIterations: 5
+    task:
+      id: output_values
+      type: io.kestra.core.tasks.outputs.OutputValues
+      values:
+        count: "{{ outputs.waitfor.iterationCount + 1 }}"
\ No newline at end of file
diff --git a/jdbc/src/test/java/io/kestra/jdbc/runner/JdbcRunnerTest.java b/jdbc/src/test/java/io/kestra/jdbc/runner/JdbcRunnerTest.java
index e2d540dd3..180764b2b 100644
--- a/jdbc/src/test/java/io/kestra/jdbc/runner/JdbcRunnerTest.java
+++ b/jdbc/src/test/java/io/kestra/jdbc/runner/JdbcRunnerTest.java
@@ -28,7 +28,6 @@ import java.util.concurrent.TimeoutException;
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.*;
-import static org.hamcrest.Matchers.is;
 
 @MicronautTest(transactional = false)
 @TestInstance(TestInstance.Lifecycle.PER_CLASS) // must be per-class to allow calling once init() which took a lot of time
@@ -76,6 +75,9 @@ public abstract class JdbcRunnerTest {
     @Inject
     private ForEachItemCaseTest forEachItemCaseTest;
 
+    @Inject
+    private WaitForCaseTest waitForTestCaseTest;
+
     @Inject
     private FlowConcurrencyCaseTest flowConcurrencyCaseTest;
 
@@ -324,4 +326,24 @@ public abstract class JdbcRunnerTest {
 
         assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
     }
+
+    @Test
+    void waitFor() throws TimeoutException {
+        waitForTestCaseTest.waitfor();
+    }
+
+    @Test
+    void waitforMaxIterations() throws TimeoutException {
+        waitForTestCaseTest.waitforMaxIterations();
+    }
+
+    @Test
+    void waitforMaxDuration() throws TimeoutException {
+        waitForTestCaseTest.waitforMaxDuration();
+    }
+
+    @Test
+    void waitforNoSuccess() throws TimeoutException {
+        waitForTestCaseTest.waitforNoSuccess();
+    }
 }