Repository: kestra-io/kestra (java)

Issue:
Odd Gantt chart after restarting an execution featuring Pause
### Expected Behavior

_No response_

### Actual Behaviour

Restarting a failed execution featuring a failing task within the `io.kestra.core.tasks.flows.Pause`  task makes the Gantt chart duplicate the `io.kestra.core.tasks.flows.Pause` task. The task seems to be somehow stuck (doesn't fail/complete).

![image](https://github.com/kestra-io/kestra/assets/13468636/580242b6-41b9-4c61-9f1f-d180bbb8bd0a)

### Steps To Reproduce

1. Save the included flow
2. Run the flow - it fails after 5s of being paused
3. Restart the failed flow
4. Check the Gantt chart

### Environment Information

- Kestra Version: 0.11.0
- Operating System (OS / Docker / Kubernetes): AL2023
- Java Version (If not docker): 17


### Example flow

```
id: pause
namespace: io.kestra.tests

tasks:
  - id: hello
    type: io.kestra.core.tasks.log.Log
    message: Pre-pause

  - id: wait-for-something
    type: io.kestra.core.tasks.flows.Pause
    delay: PT5S
    tasks:
      - id: fail-here
        type: io.kestra.core.tasks.executions.Fail

      - id: hello-post-pause
        type: io.kestra.core.tasks.log.Log
        message: Post-Pause
```

Pull request:
fix/replay-after-failing-task-under-pause
closes #2084

Patch:
diff --git a/core/src/main/java/io/kestra/core/tasks/flows/Pause.java b/core/src/main/java/io/kestra/core/tasks/flows/Pause.java
index c12f5b9c9..88c0712d3 100644
--- a/core/src/main/java/io/kestra/core/tasks/flows/Pause.java
+++ b/core/src/main/java/io/kestra/core/tasks/flows/Pause.java
@@ -123,7 +123,7 @@ public class Pause extends Sequential implements FlowableTask<VoidOutput> {
 
     private boolean needPause(TaskRun parentTaskRun) {
         return parentTaskRun.getState().getCurrent() == State.Type.RUNNING &&
-            parentTaskRun.getState().getHistories().get(parentTaskRun.getState().getHistories().size() - 2).getState() != State.Type.PAUSED;
+            parentTaskRun.getState().getHistories().stream().noneMatch(history -> history.getState() == State.Type.PAUSED);
     }
 
     @Override


Test patch:
diff --git a/core/src/test/resources/flows/valids/restart_pause_last_failed.yaml b/core/src/test/resources/flows/valids/restart_pause_last_failed.yaml
new file mode 100644
index 000000000..5072e532b
--- /dev/null
+++ b/core/src/test/resources/flows/valids/restart_pause_last_failed.yaml
@@ -0,0 +1,20 @@
+id: restart_pause_last_failed
+namespace: io.kestra.tests
+
+tasks:
+- id: a
+  type: io.kestra.core.tasks.log.Log
+  message: "{{ task.id }}"
+- id: b
+  type: io.kestra.core.tasks.log.Log
+  message: "{{ task.id }}"
+- id: pause
+  type: io.kestra.core.tasks.flows.Pause
+  delay: PT1S
+  tasks:
+  - id: c
+    type: io.kestra.core.tasks.log.Log
+    message: "{{taskrun.attemptsCount == 1 ? 'ok' : ko}}"
+  - id: d
+    type: io.kestra.core.tasks.log.Log
+    message: "{{ task.id }}"
\ No newline at end of file
diff --git a/webserver/src/test/java/io/kestra/webserver/controllers/ExecutionControllerTest.java b/webserver/src/test/java/io/kestra/webserver/controllers/ExecutionControllerTest.java
index c614d049f..c6e78cb59 100644
--- a/webserver/src/test/java/io/kestra/webserver/controllers/ExecutionControllerTest.java
+++ b/webserver/src/test/java/io/kestra/webserver/controllers/ExecutionControllerTest.java
@@ -438,6 +438,72 @@ class ExecutionControllerTest extends JdbcH2ControllerTest {
             .forEach(state -> assertThat(state.getCurrent(), is(State.Type.SUCCESS)));
     }
 
+    @Test
+    void restartFromLastFailedWithPause() throws TimeoutException {
+        final String flowId = "restart_pause_last_failed";
+
+        // Run execution until it ends
+        Execution firstExecution = runnerUtils.runOne(null, TESTS_FLOW_NS, flowId, null, null);
+
+        assertThat(firstExecution.getTaskRunList().get(2).getState().getCurrent(), is(State.Type.FAILED));
+        assertThat(firstExecution.getState().getCurrent(), is(State.Type.FAILED));
+
+        // Update task's command to make second execution successful
+        Optional<Flow> flow = flowRepositoryInterface.findById(null, TESTS_FLOW_NS, flowId);
+        assertThat(flow.isPresent(), is(true));
+
+        // Restart execution and wait until it finishes
+        Execution finishedRestartedExecution = runnerUtils.awaitExecution(
+            execution -> execution.getId().equals(firstExecution.getId()) &&
+                execution.getTaskRunList().size() == 5 &&
+                execution.getState().isTerminated(),
+            () -> {
+                Execution restartedExec = client.toBlocking().retrieve(
+                    HttpRequest
+                        .POST("/api/v1/executions/" + firstExecution.getId() + "/restart", ImmutableMap.of()),
+                    Execution.class
+                );
+
+                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));
+
+                IntStream
+                    .range(0, 2)
+                    .mapToObj(value -> restartedExec.getTaskRunList().get(value)).forEach(taskRun -> {
+                        assertThat(taskRun.getState().getCurrent(), is(State.Type.SUCCESS));
+                        assertThat(taskRun.getAttempts().size(), is(1));
+
+                        assertThat(restartedExec.getTaskRunList().get(2).getState().getCurrent(), is(State.Type.RUNNING));
+                        assertThat(restartedExec.getTaskRunList().get(3).getState().getCurrent(), is(State.Type.RESTARTED));
+                        assertThat(restartedExec.getTaskRunList().get(2).getAttempts(), nullValue());
+                        assertThat(restartedExec.getTaskRunList().get(3).getAttempts().size(), is(1));
+                    });
+            },
+            Duration.ofSeconds(15)
+        );
+
+        assertThat(finishedRestartedExecution, notNullValue());
+        assertThat(finishedRestartedExecution.getId(), is(firstExecution.getId()));
+        assertThat(finishedRestartedExecution.getParentId(), nullValue());
+        assertThat(finishedRestartedExecution.getTaskRunList().size(), is(5));
+
+        assertThat(finishedRestartedExecution.getTaskRunList().get(0).getAttempts().size(), is(1));
+        assertThat(finishedRestartedExecution.getTaskRunList().get(1).getAttempts().size(), is(1));
+        assertThat(finishedRestartedExecution.getTaskRunList().get(2).getAttempts(), nullValue());
+        assertThat(finishedRestartedExecution.getTaskRunList().get(2).getState().getHistories().stream().filter(state -> state.getState() == State.Type.PAUSED).count(), is(1L));
+        assertThat(finishedRestartedExecution.getTaskRunList().get(3).getAttempts().size(), is(2));
+        assertThat(finishedRestartedExecution.getTaskRunList().get(4).getAttempts().size(), is(1));
+
+        finishedRestartedExecution
+            .getTaskRunList()
+            .stream()
+            .map(TaskRun::getState)
+            .forEach(state -> assertThat(state.getCurrent(), is(State.Type.SUCCESS)));
+    }
+
     @Test
     void downloadFile() throws TimeoutException {
         Execution execution = runnerUtils.runOne(null, TESTS_FLOW_NS, "inputs", null, (flow, execution1) -> runnerUtils.typedInputs(flow, execution1, inputs));