fix(core): avoid restarting errors task (#1056)

close #1032
kestra-io · Mar 10, 2023 · 922e5a2 · 922e5a2
1 parent c78cb93
commit 922e5a2
Show file tree

Hide file tree

Showing 8 changed files with 147 additions and 10 deletions.
diff --git a/core/src/main/java/io/kestra/core/models/flows/Flow.java b/core/src/main/java/io/kestra/core/models/flows/Flow.java
@@ -30,7 +30,6 @@
 import javax.validation.Valid;
 import javax.validation.constraints.*;
 
-
 @SuperBuilder(toBuilder = true)
 @Getter
 @AllArgsConstructor
@@ -171,6 +170,19 @@ private Stream<Task> allTasksWithChilds(Task task) {
         }
     }
 
+    public List<Task> allErrorsWithChilds() {
+        var allErrors = allTasksWithChilds().stream()
+            .filter(task -> task.isFlowable() && ((FlowableTask<?>) task).getErrors() != null)
+            .flatMap(task -> ((FlowableTask<?>) task).getErrors().stream())
+            .collect(Collectors.toCollection(ArrayList::new));
+
+        if (this.getErrors() != null && !this.getErrors().isEmpty()) {
+            allErrors.addAll(this.getErrors());
+        }
+
+        return allErrors;
+    }
+
     public Task findTaskByTaskId(String taskId) throws InternalException {
         return allTasks()
             .flatMap(t -> t.findById(taskId).stream())

diff --git a/core/src/main/java/io/kestra/core/services/ExecutionService.java b/core/src/main/java/io/kestra/core/services/ExecutionService.java
@@ -62,7 +62,6 @@ public Execution restart(final Execution execution, @Nullable Integer revision)
 
         Set<String> taskRunToRestart = this.taskRunToRestart(
             execution,
-            flow,
             taskRun -> taskRun.getState().getCurrent().isFailed()
         );
 
@@ -86,6 +85,11 @@ public Execution restart(final Execution execution, @Nullable Integer revision)
         this.removeWorkerTask(flow, execution, taskRunToRestart, mappingTaskRunId)
             .forEach(r -> newTaskRuns.removeIf(taskRun -> taskRun.getId().equals(r)));
 
+        // We need to remove global error tasks and flowable error tasks if any
+        flow
+            .allErrorsWithChilds()
+            .forEach(task -> newTaskRuns.removeIf(taskRun -> taskRun.getTaskId().equals(task.getId())));
+
         // Build and launch new execution
         Execution newExecution = execution
             .childExecution(
@@ -97,7 +101,7 @@ public Execution restart(final Execution execution, @Nullable Integer revision)
         return revision != null ? newExecution.withFlowRevision(revision) : newExecution;
     }
 
-    private Set<String> taskRunToRestart(Execution execution, Flow flow, Predicate<TaskRun> predicate) throws InternalException {
+    private Set<String> taskRunToRestart(Execution execution, Predicate<TaskRun> predicate) {
         // Original tasks to be restarted
         Set<String> finalTaskRunToRestart = this
             .taskRunWithAncestors(
@@ -128,7 +132,6 @@ public Execution replay(final Execution execution, String taskRunId, @Nullable I
 
         Set<String> taskRunToRestart = this.taskRunToRestart(
             execution,
-            flow,
             taskRun -> taskRun.getId().equals(taskRunId)
         );
 
@@ -187,7 +190,6 @@ public Execution markAs(final Execution execution, String taskRunId, State.Type
 
         Set<String> taskRunToRestart = this.taskRunToRestart(
             execution,
-            flow,
             taskRun -> taskRun.getId().equals(taskRunId)
         );
 

diff --git a/core/src/test/java/io/kestra/core/Helpers.java b/core/src/test/java/io/kestra/core/Helpers.java
@@ -19,7 +19,7 @@
 import java.util.function.Consumer;
 
 public class Helpers {
-    public static long FLOWS_COUNT = 53;
+    public static long FLOWS_COUNT = 55;
 
     public static ApplicationContext applicationContext() throws URISyntaxException {
         return applicationContext(

diff --git a/core/src/test/java/io/kestra/core/runners/RestartCaseTest.java b/core/src/test/java/io/kestra/core/runners/RestartCaseTest.java
@@ -34,7 +34,7 @@ public class RestartCaseTest {
     @Named(QueueFactoryInterface.EXECUTION_NAMED)
     private QueueInterface<Execution> executionQueue;
 
-    public void restartFailed() throws Exception {
+    public void restartFailedThenSuccess() throws Exception {
         Flow flow = flowRepository.findById("io.kestra.tests", "restart_last_failed").orElseThrow();
 
         Execution firstExecution = runnerUtils.runOne(flow.getNamespace(), flow.getId(), Duration.ofSeconds(60));
@@ -74,6 +74,78 @@ public void restartFailed() throws Exception {
             .forEach(state -> assertThat(state.getCurrent(), is(State.Type.SUCCESS)));
     }
 
+    public void restartFailedThenFailureWithGlobalErrors() throws Exception {
+        Flow flow = flowRepository.findById("io.kestra.tests", "restart_always_failed").orElseThrow();
+
+        Execution firstExecution = runnerUtils.runOne(flow.getNamespace(), flow.getId(), Duration.ofSeconds(60));
+
+        assertThat(firstExecution.getState().getCurrent(), is(State.Type.FAILED));
+        assertThat(firstExecution.getTaskRunList(), hasSize(2));
+        assertThat(firstExecution.getTaskRunList().get(0).getState().getCurrent(), is(State.Type.FAILED));
+
+        // wait
+        Execution finishedRestartedExecution = runnerUtils.awaitExecution(
+            execution -> execution.getState().getCurrent() == State.Type.FAILED,
+            throwRunnable(() -> {
+                Thread.sleep(1000);
+                Execution restartedExec = executionService.restart(firstExecution, null);
+                executionQueue.emit(restartedExec);
+
+                assertThat(restartedExec, notNullValue());
+                assertThat(restartedExec.getId(), is(firstExecution.getId()));
+                assertThat(restartedExec.getParentId(), nullValue());
+                assertThat(restartedExec.getTaskRunList().size(), is(1));
+                assertThat(restartedExec.getState().getCurrent(), is(State.Type.RESTARTED));
+            }),
+            Duration.ofSeconds(60)
+        );
+
+        assertThat(finishedRestartedExecution, notNullValue());
+        assertThat(finishedRestartedExecution.getId(), is(firstExecution.getId()));
+        assertThat(finishedRestartedExecution.getParentId(), nullValue());
+        assertThat(finishedRestartedExecution.getTaskRunList().size(), is(2));
+
+        assertThat(finishedRestartedExecution.getTaskRunList().get(0).getAttempts().size(), is(2));
+
+        assertThat(finishedRestartedExecution.getState().getCurrent(), is(State.Type.FAILED));
+    }
+
+    public void restartFailedThenFailureWithLocalErrors() throws Exception {
+        Flow flow = flowRepository.findById("io.kestra.tests", "restart_local_errors").orElseThrow();
+
+        Execution firstExecution = runnerUtils.runOne(flow.getNamespace(), flow.getId(), Duration.ofSeconds(60));
+
+        assertThat(firstExecution.getState().getCurrent(), is(State.Type.FAILED));
+        assertThat(firstExecution.getTaskRunList(), hasSize(5));
+        assertThat(firstExecution.getTaskRunList().get(3).getState().getCurrent(), is(State.Type.FAILED));
+
+        // wait
+        Execution finishedRestartedExecution = runnerUtils.awaitExecution(
+            execution -> execution.getState().getCurrent() == State.Type.FAILED,
+            throwRunnable(() -> {
+                Thread.sleep(1000);
+                Execution restartedExec = executionService.restart(firstExecution, null);
+                executionQueue.emit(restartedExec);
+
+                assertThat(restartedExec, notNullValue());
+                assertThat(restartedExec.getId(), is(firstExecution.getId()));
+                assertThat(restartedExec.getParentId(), nullValue());
+                assertThat(restartedExec.getTaskRunList().size(), is(4));
+                assertThat(restartedExec.getState().getCurrent(), is(State.Type.RESTARTED));
+            }),
+            Duration.ofSeconds(60)
+        );
+
+        assertThat(finishedRestartedExecution, notNullValue());
+        assertThat(finishedRestartedExecution.getId(), is(firstExecution.getId()));
+        assertThat(finishedRestartedExecution.getParentId(), nullValue());
+        assertThat(finishedRestartedExecution.getTaskRunList().size(), is(5));
+
+        assertThat(finishedRestartedExecution.getTaskRunList().get(3).getAttempts().size(), is(2));
+
+        assertThat(finishedRestartedExecution.getState().getCurrent(), is(State.Type.FAILED));
+    }
+
     public void replay() throws Exception {
         Flow flow = flowRepository.findById("io.kestra.tests", "restart-each").orElseThrow();
 

diff --git a/core/src/test/java/io/kestra/core/runners/RestartTest.java b/core/src/test/java/io/kestra/core/runners/RestartTest.java
@@ -9,8 +9,18 @@ public class RestartTest extends AbstractMemoryRunnerTest {
     private RestartCaseTest restartCaseTest;
 
     @Test
-    void restartFailed() throws Exception {
-        restartCaseTest.restartFailed();
+    void restartFailedThenSuccess() throws Exception {
+        restartCaseTest.restartFailedThenSuccess();
+    }
+
+    @Test
+    void restartFailedThenFailureWithGlobalErrors() throws Exception {
+        restartCaseTest.restartFailedThenFailureWithGlobalErrors();
+    }
+
+    @Test
+    void restartFailedThenFailureWithLocalErrors() throws Exception {
+        restartCaseTest.restartFailedThenFailureWithLocalErrors();
     }
 
     @Test

diff --git a/core/src/test/resources/flows/valids/restart_always_failed.yaml b/core/src/test/resources/flows/valids/restart_always_failed.yaml
@@ -0,0 +1,14 @@
+id: restart_always_failed
+namespace: io.kestra.tests
+
+tasks:
+  - id: failStep
+    type: io.kestra.core.tasks.scripts.Bash
+    description: "This fails"
+    commands:
+      - 'exit 1'
+errors:
+  - id: errorHandler
+    type: io.kestra.core.tasks.debugs.Echo
+    format: I'm failing {{task.id}}
+    level: INFO
diff --git a/core/src/test/resources/flows/valids/restart_local_errors.yaml b/core/src/test/resources/flows/valids/restart_local_errors.yaml
@@ -0,0 +1,27 @@
+id: restart_local_errors
+namespace: io.kestra.tests
+
+tasks:
+  - id: before
+    type: io.kestra.core.tasks.debugs.Echo
+    format: I'm before
+
+  - id: sequential
+    type: io.kestra.core.tasks.flows.Sequential
+    tasks:
+    - id: close
+      type: io.kestra.core.tasks.debugs.Echo
+      format: I'm close to fail
+    - id: failStep
+      type: io.kestra.core.tasks.scripts.Bash
+      description: "This fails"
+      commands:
+        - 'exit 1'
+    errors:
+    - id: errorHandler
+      type: io.kestra.core.tasks.debugs.Echo
+      format: I'm failing {{task.id}}
+
+  - id: after
+    type: io.kestra.core.tasks.debugs.Echo
+    format: I'm after
diff --git a/jdbc/src/test/java/io/kestra/jdbc/runner/JdbcRunnerTest.java b/jdbc/src/test/java/io/kestra/jdbc/runner/JdbcRunnerTest.java
@@ -143,7 +143,7 @@ void eachParallelNested() throws TimeoutException {
 
     @Test
     void restartFailed() throws Exception {
-        restartCaseTest.restartFailed();
+        restartCaseTest.restartFailedThenSuccess();
     }
 
     @Test
-Original file line number
+Diff line change
@@ Expand Up / @@ -143,7 +143,7 @@ void eachParallelNested() throws TimeoutException { @@
         @Test
         void restartFailed() throws Exception {
-            restartCaseTest.restartFailed();
+            restartCaseTest.restartFailedThenSuccess();
         }
         @Test
@@ Expand Down @@