fix(core): killing paused without subtask should transition to KILLED

Fixes #6243 When we kill an execution that is running a Pause task that didn't have any subtask, we must transition the task run to KILLED immediatly or the executor will process the Pause task and transition it to SUCCESS.
kestra-io · Jan 2, 2025 · a9ff469 · a9ff469
1 parent 6db49ae
commit a9ff469
Show file tree

Hide file tree

Showing 2 changed files with 25 additions and 4 deletions.
diff --git a/core/src/main/java/io/kestra/core/services/ExecutionService.java b/core/src/main/java/io/kestra/core/services/ExecutionService.java
@@ -322,13 +322,18 @@ private Execution markAs(final Execution execution, Flow flow, String taskRunId,
                     newTaskRun = newTaskRun.withOutputs(pauseTask.generateOutputs(onResumeInputs));
                 }
 
-                if (task instanceof Pause pauseTask && pauseTask.getTasks() == null && newState == State.Type.RUNNING) {
-                    newTaskRun = newTaskRun.withState(State.Type.SUCCESS);
+                // if it's a Pause task with no subtask, we terminate the task
+                if (task instanceof Pause pauseTask && pauseTask.getTasks() == null) {
+                    if (newState == State.Type.RUNNING) {
+                        newTaskRun = newTaskRun.withState(State.Type.SUCCESS);
+                    } else if (newState == State.Type.KILLING) {
+                        newTaskRun = newTaskRun.withState(State.Type.KILLED);
+                    }
                 }
 
                 if (originalTaskRun.getAttempts() != null && !originalTaskRun.getAttempts().isEmpty()) {
                     ArrayList<TaskRunAttempt> attempts = new ArrayList<>(originalTaskRun.getAttempts());
-                    attempts.set(attempts.size() - 1, attempts.get(attempts.size() - 1).withState(newState));
+                    attempts.set(attempts.size() - 1, attempts.getLast().withState(newState));
                     newTaskRun = newTaskRun.withAttempts(attempts);
                 }
 
@@ -450,7 +455,7 @@ public void delete(
      * The execution must be paused or this call will be a no-op.
      *
      * @param execution the execution to resume
-     * @param newState  should be RUNNING or KILLING, other states may lead to undefined behaviour
+     * @param newState  should be RUNNING or KILLING, other states may lead to undefined behavior
      * @param flow      the flow of the execution
      * @return the execution in the new state.
      * @throws Exception if the state of the execution cannot be updated

diff --git a/core/src/test/java/io/kestra/core/runners/ExecutionServiceTest.java b/core/src/test/java/io/kestra/core/runners/ExecutionServiceTest.java
@@ -387,4 +387,20 @@ void deleteExecutionKeepLogs(Execution execution) throws IOException {
         assertThat(executionRepository.findById(execution.getTenantId(),execution.getId()), is(Optional.empty()));
         assertThat(logRepository.findByExecutionId(execution.getTenantId(),execution.getId(), Level.INFO), hasSize(4));
     }
+
+    @Test
+    @LoadFlows({"flows/valids/pause_no_tasks.yaml"})
+    void shouldKillPausedExecutions() throws Exception {
+        Execution execution = runnerUtils.runOneUntilPaused(null, "io.kestra.tests", "pause_no_tasks");
+        Flow flow = flowRepository.findByExecution(execution);
+
+        assertThat(execution.getTaskRunList(), hasSize(1));
+        assertThat(execution.getState().getCurrent(), is(State.Type.PAUSED));
+
+        Execution killed = executionService.kill(execution, flow);
+
+        assertThat(killed.getState().getCurrent(), is(State.Type.RESTARTED));
+        assertThat(killed.findTaskRunsByTaskId("pause").getFirst().getState().getCurrent(), is(State.Type.KILLED));
+        assertThat(killed.getState().getHistories(), hasSize(4));
+    }
 }