Repository: kestra-io/kestra (java)

Issue:
Add Feature for Pause Plugin to Cancel/Kill After Timeout is Reached
### Feature description

Current the [Pause Plugin](https://kestra.io/plugins/core/flow/io.kestra.plugin.core.flow.pause#properties_timeout-body) has a `Timeout` property that `RESUMES` the Workflow after it is reached. I am proposing a feature to have it `CANCEL` or `KILL` the Workflow when it is reached.

Reasoning: Currently, we have some Workflows that run daily and require manual approval to continue. If a user misses it over a weekend, or holiday before timeout is reached, I want it to `CANCEL/KILL` instead of auto-resuming since I want these processes to always have a manual check through.

Is this possible and reasonable?

Let me know if there are questions, context, or if you would like a dialogue.

Thank you team as always 🚀 🤖 🪂 

Pull request description:
feat(core): allow to specify Pause task resume behavior (#8242)
When a pause task is resumed either because it wait until the end of it's duration or is resumed manually, it can now use a behavior to describe what to do next: resume, warn, fail, or cancel the execution.

Fixes #8242

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 46de05370..ae776570c 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
@@ -69,7 +69,7 @@ public class State {
 
     public State withState(Type state) {
         if (this.current == state) {
-            log.warn("Can't change state, already " + current);
+            log.warn("Can't change state, already {}", current);
             return this;
         }
 
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 a42135988..494e697e5 100644
--- a/core/src/main/java/io/kestra/core/runners/ExecutorService.java
+++ b/core/src/main/java/io/kestra/core/runners/ExecutorService.java
@@ -615,16 +615,19 @@ public class ExecutorService {
                 Task task = executor.getFlow().findTaskByTaskId(workerTaskResult.getTaskRun().getTaskId());
 
                 if (task instanceof Pause pauseTask) {
-                    if (pauseTask.getDelay() != null || pauseTask.getTimeout() != null) {
+                    if (pauseTask.getPauseDuration() != null || pauseTask.getTimeout() != null) {
                         RunContext runContext = runContextFactory.of(executor.getFlow(), executor.getExecution());
-                        Duration delay = runContext.render(pauseTask.getDelay()).as(Duration.class).orElse(null);
+                        Duration duration = runContext.render(pauseTask.getPauseDuration()).as(Duration.class).orElse(null);
                         Duration timeout = runContext.render(pauseTask.getTimeout()).as(Duration.class).orElse(null);
-                        if (delay != null || timeout != null) { // rendering can lead to null, so we must re-check here
+                        Pause.Behavior behavior  = runContext.render(pauseTask.getBehavior()).as(Pause.Behavior.class).orElse(Pause.Behavior.RESUME);
+                        if (duration != null || timeout != null) { // rendering can lead to null, so we must re-check here
+                            // if duration is set, we use it, and we use the Pause behavior as a state
+                            // if no duration, we use the standard timeout property and use FAILED as the target state
                             return ExecutionDelay.builder()
                                 .taskRunId(workerTaskResult.getTaskRun().getId())
                                 .executionId(executor.getExecution().getId())
-                                .date(workerTaskResult.getTaskRun().getState().maxDate().plus(delay != null ? delay : timeout))
-                                .state(delay != null ? State.Type.RUNNING : State.Type.FAILED)
+                                .date(workerTaskResult.getTaskRun().getState().maxDate().plus(duration != null ? duration : timeout))
+                                .state(duration != null ? behavior.mapToState() : State.Type.FAILED)
                                 .delayType(ExecutionDelay.DelayType.RESUME_FLOW)
                                 .build();
                         }
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 b92985a52..1f6f7a9f8 100644
--- a/core/src/main/java/io/kestra/core/services/ExecutionService.java
+++ b/core/src/main/java/io/kestra/core/services/ExecutionService.java
@@ -344,7 +344,7 @@ public class ExecutionService {
                 }
 
                 // if it's a Pause task with no subtask, we terminate the task
-                if (task instanceof Pause pauseTask && pauseTask.getTasks() == null) {
+                if (task instanceof Pause pauseTask && ListUtils.isEmpty(pauseTask.getTasks())) {
                     if (newState == State.Type.RUNNING) {
                         newTaskRun = newTaskRun.withState(State.Type.SUCCESS);
                     } else if (newState == State.Type.KILLING) {
@@ -365,11 +365,12 @@ public class ExecutionService {
         }
 
         if (newExecution.getTaskRunList().stream().anyMatch(t -> t.getState().getCurrent() == State.Type.PAUSED)) {
-            // there is still some tasks paused, this can occur with parallel pause
+            // there are still some tasks paused, this can occur with parallel pause
             return newExecution;
         }
-        return newExecution
-            .withState(State.Type.RESTARTED);
+
+        // we need to cancel immediately or the executor will process the next task if it's restarted.
+        return newState == State.Type.CANCELLED ? newExecution.withState(State.Type.CANCELLED) : newExecution.withState(State.Type.RESTARTED);
     }
 
     public Execution markWithTaskRunAs(final Execution execution, String taskRunId, State.Type newState, Boolean markParents) throws Exception {
@@ -655,7 +656,7 @@ public class ExecutionService {
      *
      * @return the execution in a KILLING state if not already terminated
      */
-    public Execution kill(Execution execution, Flow flow) {
+    public Execution kill(Execution execution, FlowInterface flow) {
         if (execution.getState().getCurrent() == State.Type.KILLING || execution.getState().isTerminated()) {
             return execution;
         }
diff --git a/core/src/main/java/io/kestra/plugin/core/flow/Pause.java b/core/src/main/java/io/kestra/plugin/core/flow/Pause.java
index 651588d97..dc571ee50 100644
--- a/core/src/main/java/io/kestra/plugin/core/flow/Pause.java
+++ b/core/src/main/java/io/kestra/plugin/core/flow/Pause.java
@@ -22,8 +22,10 @@ import io.kestra.core.runners.FlowableUtils;
 import io.kestra.core.runners.RunContext;
 import io.kestra.core.serializers.JacksonMapper;
 import io.kestra.core.utils.GraphUtils;
+import io.kestra.core.utils.ListUtils;
 import io.swagger.v3.oas.annotations.media.Schema;
 import jakarta.validation.Valid;
+import jakarta.validation.constraints.NotNull;
 import lombok.*;
 import lombok.experimental.SuperBuilder;
 
@@ -40,9 +42,9 @@ import java.util.stream.Stream;
 @Getter
 @NoArgsConstructor
 @Schema(
-    title = "Pause the current execution and wait for approval (either by humans or other automated processes).", 
-    description = "All tasks downstream from the Pause task will be put on hold until the execution is manually resumed from the UI.\n\n" + 
-      "The Execution will be in a Paused state, and you can either manually resume it by clicking on the \"Resume\" button in the UI or by calling the POST API endpoint `/api/v1/executions/{executionId}/resume`. The execution can also be resumed automatically after a timeout."
+    title = "Pause the current execution and wait for approval (either by humans or other automated processes).",
+    description = "All tasks downstream from the Pause task will be put on hold until the execution is manually resumed from the UI.\n\n" +
+      "The Execution will be in a Paused state, and you can either manually resume it by clicking on the \"Resume\" button in the UI or by calling the POST API endpoint `/api/v1/executions/{executionId}/resume`. The execution can also be resumed automatically after the `pauseDuration`."
 )
 @Plugin(
     examples = {
@@ -129,6 +131,24 @@ import java.util.stream.Stream;
                     type: io.kestra.plugin.core.log.Log
                     message: Status is {{ outputs.wait_for_approval.onResume.reason }}. Process finished with {{ outputs.approve.body }}
                 """
+        ),
+        @Example(
+            title = "Pause the execution and set the execution in WARNING if it has not been resumed after 5 minutes",
+            full = true,
+            code = """
+                id: pause_warn
+                namespace: company.team
+
+                tasks:
+                  - id: pause
+                    type: io.kestra.plugin.core.flow.Pause
+                    pauseDuration: PT5M
+                    behavior: WARN
+
+                  - id: post_resume
+                    type: io.kestra.plugin.core.debug.Return
+                    format: "{{ task.id }} started on {{ taskrun.startDate }} after the Pause"
+                """
         )
     },
     aliases = "io.kestra.core.tasks.flows.Pause"
@@ -136,17 +156,38 @@ import java.util.stream.Stream;
 public class Pause extends Task implements FlowableTask<Pause.Output> {
     @Schema(
         title = "Duration of the pause — useful if you want to pause the execution for a fixed amount of time.",
-        description = "The delay is a string in the [ISO 8601 Duration](https://en.wikipedia.org/wiki/ISO_8601#Durations) format, e.g. `PT1H` for 1 hour, `PT30M` for 30 minutes, `PT10S` for 10 seconds, `P1D` for 1 day, etc. If no delay and no timeout are configured, the execution will never end until it's manually resumed from the UI or API.",
+        description = "**Deprecated**: use `pauseDuration` instead.",
         implementation = Duration.class
     )
+    @Deprecated
     private Property<Duration> delay;
 
+    @Deprecated
+    public void setDelay(Property<Duration> delay) {
+        this.delay = delay;
+        this.pauseDuration = delay;
+    }
+
     @Schema(
-        title = "Timeout of the pause — useful to avoid never-ending workflows in a human-in-the-loop scenario. For example, if you want to pause the execution until a human validates some data generated in a previous task, you can set a timeout of e.g. 24 hours. If no manual approval happens within 24 hours, the execution will automatically resume without a prior data validation.",
-        description = "If no delay and no timeout are configured, the execution will never end until it's manually resumed from the UI or API.",
+        title = "Duration of the pause. If not set the task will wait forever to be manually resumed except if a timeout is set, in this case, the timeout will be honored.",
+        description = "The duration is a string in the [ISO 8601 Duration](https://en.wikipedia.org/wiki/ISO_8601#Durations) format, e.g. `PT1H` for 1 hour, `PT30M` for 30 minutes, `PT10S` for 10 seconds, `P1D` for 1 day, etc. If no pauseDuration and no timeout are configured, the execution will never end until it's manually resumed from the UI or API.",
         implementation = Duration.class
     )
-    private Property<Duration> timeout;
+    private Property<Duration> pauseDuration;
+
+    @Schema(
+        title = "Pause behavior, by default RESUME. What happens when a pause task reach its duration.",
+        description = """
+            Tasks that are resumed before the duration (for example, from the UI) will not use the behavior property but will always success.
+            Possible values are:
+            - RESUME: continue with the execution
+            - WARN: ends the Pause task in WARNING and continue with the execution
+            - FAIL: fail the Pause task
+            - CANCEL: cancel the execution"""
+    )
+    @NotNull
+    @Builder.Default
+    private Property<Behavior> behavior = Property.of(Behavior.RESUME);
 
     @Valid
     @Schema(
@@ -230,17 +271,34 @@ public class Pause extends Task implements FlowableTask<Pause.Output> {
             parentTaskRun.getState().getHistories().stream().noneMatch(history -> history.getState() == State.Type.PAUSED);
     }
 
+    // This method is only called when there are subtasks
     @Override
     public Optional<State.Type> resolveState(RunContext runContext, Execution execution, TaskRun parentTaskRun) throws IllegalVariableEvaluationException {
         if (this.needPause(parentTaskRun)) {
             return Optional.of(State.Type.PAUSED);
         }
 
-        if (this.tasks == null || this.tasks.isEmpty()) {
-            return Optional.of(State.Type.SUCCESS);
-        }
-
-        return FlowableTask.super.resolveState(runContext, execution, parentTaskRun);
+        Behavior behavior  = runContext.render(this.behavior).as(Behavior.class).orElse(Behavior.RESUME);
+        return switch (behavior) {
+            case Behavior.RESUME -> {
+                // yield SUCCESS or the final flowable task state
+                if (ListUtils.isEmpty(this.tasks)) {
+                    yield Optional.of(State.Type.SUCCESS);
+                } else {
+                    yield FlowableTask.super.resolveState(runContext, execution, parentTaskRun);
+                }
+            }
+            case Behavior.WARN -> {
+                // yield WARNING or the final flowable task state, if the flowable ends in SUCCESS, yield WARNING
+                if (ListUtils.isEmpty(this.tasks)) {
+                    yield Optional.of(State.Type.WARNING);
+                } else {
+                    Optional<State.Type> finalState = FlowableTask.super.resolveState(runContext, execution, parentTaskRun);
+                    yield finalState.map(state -> state == State.Type.SUCCESS ? State.Type.WARNING : state);
+                }
+            }
+            case Behavior.CANCEL ,Behavior.FAIL -> throw new IllegalArgumentException("The " + behavior + " cannot be handled at this stage, this is certainly a bug!");
+        };
     }
 
     public Map<String, Object> generateOutputs(Map<String, Object> inputs) {
@@ -256,4 +314,21 @@ public class Pause extends Task implements FlowableTask<Pause.Output> {
     public static class Output implements io.kestra.core.models.tasks.Output {
         private Map<String, Object> onResume;
     }
+
+    public enum Behavior {
+        RESUME(State.Type.RUNNING),
+        WARN(State.Type.WARNING),
+        CANCEL(State.Type.CANCELLED),
+        FAIL(State.Type.FAILED);
+
+        private final State.Type executionState;
+
+        Behavior(State.Type executionState) {
+            this.executionState = executionState;
+        }
+
+        public State.Type mapToState() {
+            return this.executionState;
+        }
+    }
 }


Test patch:
diff --git a/core/src/test/java/io/kestra/core/runners/AbstractRunnerTest.java b/core/src/test/java/io/kestra/core/runners/AbstractRunnerTest.java
index fee2cdeec..6a549ada2 100644
--- a/core/src/test/java/io/kestra/core/runners/AbstractRunnerTest.java
+++ b/core/src/test/java/io/kestra/core/runners/AbstractRunnerTest.java
@@ -305,9 +305,9 @@ public abstract class AbstractRunnerTest {
     }
 
     @Test
-    @LoadFlows({"flows/valids/pause-delay-from-input.yaml"})
-    public void pauseRunDelayFromInput() throws Exception {
-        pauseTest.runDelayFromInput(runnerUtils);
+    @LoadFlows({"flows/valids/pause-duration-from-input.yaml"})
+    public void pauseRunDurationFromInput() throws Exception {
+        pauseTest.runDurationFromInput(runnerUtils);
     }
 
     @Test
diff --git a/core/src/test/java/io/kestra/plugin/core/flow/PauseTest.java b/core/src/test/java/io/kestra/plugin/core/flow/PauseTest.java
index 1115e53ea..5230dcb18 100644
--- a/core/src/test/java/io/kestra/plugin/core/flow/PauseTest.java
+++ b/core/src/test/java/io/kestra/plugin/core/flow/PauseTest.java
@@ -63,7 +63,7 @@ public class PauseTest {
 
     @Disabled("This test is too flaky and it always pass in JDBC and Kafka")
     void delayFromInput() throws Exception {
-        suite.runDelayFromInput(runnerUtils);
+        suite.runDurationFromInput(runnerUtils);
     }
 
     @Disabled("This test is too flaky and it always pass in JDBC and Kafka")
@@ -101,6 +101,30 @@ public class PauseTest {
         suite.runOnResumeOptionalInputs(runnerUtils);
     }
 
+    @Test
+    @LoadFlows({"flows/valids/pause-behavior.yaml"})
+    void runDurationWithCONTINUEBehavior() throws Exception {
+        suite.runDurationWithBehavior(runnerUtils, Pause.Behavior.RESUME);
+    }
+
+    @Test
+    @LoadFlows({"flows/valids/pause-behavior.yaml"})
+    void runDurationWithFAILBehavior() throws Exception {
+        suite.runDurationWithBehavior(runnerUtils, Pause.Behavior.FAIL);
+    }
+
+    @Test
+    @LoadFlows({"flows/valids/pause-behavior.yaml"})
+    void runDurationWithWARNBehavior() throws Exception {
+        suite.runDurationWithBehavior(runnerUtils, Pause.Behavior.WARN);
+    }
+
+    @Test
+    @LoadFlows({"flows/valids/pause-behavior.yaml"})
+    void runDurationWithCANCELBehavior() throws Exception {
+        suite.runDurationWithBehavior(runnerUtils, Pause.Behavior.CANCEL);
+    }
+
     @Singleton
     public static class Suite {
         @Inject
@@ -159,8 +183,8 @@ public class PauseTest {
             assertThat(execution.getTaskRunList()).hasSize(3);
         }
 
-        public void runDelayFromInput(RunnerUtils runnerUtils) throws Exception {
-            Execution execution = runnerUtils.runOneUntilPaused(null, "io.kestra.tests", "pause-delay-from-input", null, null, Duration.ofSeconds(30));
+        public void runDurationFromInput(RunnerUtils runnerUtils) throws Exception {
+            Execution execution = runnerUtils.runOneUntilPaused(null, "io.kestra.tests", "pause-duration-from-input", null, null, Duration.ofSeconds(30));
             String executionId = execution.getId();
 
             assertThat(execution.getState().getCurrent()).isEqualTo(State.Type.PAUSED);
@@ -300,5 +324,25 @@ public class PauseTest {
             Map<String, Object> outputs = (Map<String, Object>) execution.findTaskRunsByTaskId("last").getFirst().getOutputs().get("values");
             assertThat(outputs.get("asked")).isEqualTo("MISSING");
         }
+
+        public void runDurationWithBehavior(RunnerUtils runnerUtils, Pause.Behavior behavior) throws Exception {
+            Execution execution = runnerUtils.runOneUntilPaused(null, "io.kestra.tests", "pause-behavior", null, (unused, _unused) -> Map.of("behavior", behavior), Duration.ofSeconds(30));
+            String executionId = execution.getId();
+
+            assertThat(execution.getState().getCurrent()).isEqualTo(State.Type.PAUSED);
+            assertThat(execution.getTaskRunList()).hasSize(1);
+
+            execution = runnerUtils.awaitExecution(
+                e -> e.getId().equals(executionId) && e.getState().getCurrent().isTerminated(),
+                () -> {},
+                Duration.ofSeconds(5)
+            );
+
+            State.Type finalState = behavior == Pause.Behavior.RESUME ? State.Type.SUCCESS : behavior.mapToState();
+            boolean terminateAfterPause = behavior == Pause.Behavior.CANCEL || behavior == Pause.Behavior.FAIL;
+            assertThat(execution.getTaskRunList()).hasSize(terminateAfterPause ? 1 : 2);
+            assertThat(execution.getTaskRunList().getFirst().getState().getCurrent()).isEqualTo(finalState);
+            assertThat(execution.getState().getCurrent()).isEqualTo(finalState);
+        }
     }
 }
\ No newline at end of file
diff --git a/core/src/test/resources/flows/valids/pause-behavior.yaml b/core/src/test/resources/flows/valids/pause-behavior.yaml
new file mode 100644
index 000000000..074f1ea71
--- /dev/null
+++ b/core/src/test/resources/flows/valids/pause-behavior.yaml
@@ -0,0 +1,16 @@
+id: pause-behavior
+namespace: io.kestra.tests
+
+inputs:
+  - id: behavior
+    type: STRING
+    defaults: CONTINUE
+
+tasks:
+  - id: pause
+    type: io.kestra.plugin.core.flow.Pause
+    pauseDuration: PT1S
+    behavior: "{{inputs.behavior}}"
+  - id: last
+    type: io.kestra.plugin.core.debug.Return
+    format: "{{task.id}}"
diff --git a/core/src/test/resources/flows/valids/pause-delay-from-input.yaml b/core/src/test/resources/flows/valids/pause-duration-from-input.yaml
similarity index 84%
rename from core/src/test/resources/flows/valids/pause-delay-from-input.yaml
rename to core/src/test/resources/flows/valids/pause-duration-from-input.yaml
index dff811fa1..c02cc90cc 100644
--- a/core/src/test/resources/flows/valids/pause-delay-from-input.yaml
+++ b/core/src/test/resources/flows/valids/pause-duration-from-input.yaml
@@ -1,4 +1,4 @@
-id: pause-delay-from-input
+id: pause-duration-from-input
 namespace: io.kestra.tests
 
 inputs:
@@ -9,7 +9,7 @@ inputs:
 tasks:
   - id: pause
     type: io.kestra.plugin.core.flow.Pause
-    delay: "{{inputs.delay}}"
+    pauseDuration: "{{inputs.delay}}"
     tasks:
       - id: subtask
         type: io.kestra.plugin.core.log.Log