Repository: kestra-io/kestra (java)

Issue:
Parent flow with `retry` does not retry when Subflow fails and `transmitFailed` set to true 
### Describe the issue

This parent flow will call a Subflow which will fail. `transmitFailed` is set to true meaning that the parent flow fails too. However, the parent flow does not retry despite being defined.

Parent flow:
```yaml
id: retry_flow
namespace: kestra.sandbox

retry:
  maxAttempt: 2
  behavior: CREATE_NEW_EXECUTION
  type: constant
  interval: PT2S

tasks:
  - id: call_subflow
    type: io.kestra.plugin.core.flow.Subflow
    namespace: "{{ flow.namespace }}"
    flowId: fail_flow
    wait: true
    transmitFailed: true
```
Subflow:
```yaml
id: fail_flow
namespace: kestra.sandbox

tasks:
  - id: fail
    type: io.kestra.plugin.scripts.shell.Commands
    commands:
      - exit 1
```
What happened:
![Screenshot 2024-09-04 at 13 36 39](https://github.com/user-attachments/assets/32c50f77-7e68-463f-9905-0912f9d5cd5c)

What should happen (parent flow is forced to fail separately to subflow to generate expected result):
![Screenshot 2024-09-04 at 13 37 11](https://github.com/user-attachments/assets/46678c35-cda2-470e-8035-58858e8cf3a4)



### Environment

- Kestra Version: 0.19 Snapshot as well as 0.18.4
- Operating System (OS/Docker/Kubernetes):
- Java Version (if you don't run kestra in Docker):


Pull request description:
fix(core): handle subflow retries
closes #4823


Patch:
diff --git a/core/src/main/java/io/kestra/core/runners/ExecutableUtils.java b/core/src/main/java/io/kestra/core/runners/ExecutableUtils.java
index 62e140797..db84cad02 100644
--- a/core/src/main/java/io/kestra/core/runners/ExecutableUtils.java
+++ b/core/src/main/java/io/kestra/core/runners/ExecutableUtils.java
@@ -16,11 +16,7 @@ import io.kestra.core.models.tasks.Task;
 import io.kestra.core.storages.Storage;
 import lombok.extern.slf4j.Slf4j;
 
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
+import java.util.*;
 
 @Slf4j
 public final class ExecutableUtils {
@@ -44,12 +40,12 @@ public final class ExecutableUtils {
     }
 
     public static SubflowExecutionResult subflowExecutionResult(TaskRun parentTaskrun, Execution execution) {
+        List<TaskRunAttempt> attempts = parentTaskrun.getAttempts() == null ? new ArrayList<>() : new ArrayList<>(parentTaskrun.getAttempts());
+        attempts.add(TaskRunAttempt.builder().state(parentTaskrun.getState()).build());
         return SubflowExecutionResult.builder()
             .executionId(execution.getId())
             .state(parentTaskrun.getState().getCurrent())
-            .parentTaskRun(parentTaskrun.withAttempts(
-                Collections.singletonList(TaskRunAttempt.builder().state(parentTaskrun.getState()).build())
-            ))
+            .parentTaskRun(parentTaskrun.withAttempts(attempts))
             .build();
     }
 
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 b3bdceb5d..73508341b 100644
--- a/core/src/main/java/io/kestra/core/runners/ExecutorService.java
+++ b/core/src/main/java/io/kestra/core/runners/ExecutorService.java
@@ -12,6 +12,7 @@ import io.kestra.core.services.ConditionService;
 import io.kestra.core.services.ExecutionService;
 import io.kestra.core.services.LogService;
 import io.kestra.core.utils.ListUtils;
+import io.kestra.plugin.core.flow.Subflow;
 import io.kestra.plugin.core.flow.WaitFor;
 import io.kestra.plugin.core.flow.Pause;
 import io.kestra.plugin.core.flow.WorkingDirectory;
@@ -483,7 +484,7 @@ public class ExecutorService {
              */
             if (!executor.getExecution().getState().isRetrying() &&
                 taskRun.getState().isFailed() &&
-                task instanceof RunnableTask<?> &&
+                (task instanceof RunnableTask<?> || task instanceof Subflow) &&
                 (task.getRetry() != null || executor.getFlow().getRetry() != null || (parentTask != null && parentTask.getRetry() != null))
             ) {
                 Instant nextRetryDate;
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 8f76c1af3..a2d11d133 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java
@@ -917,11 +917,11 @@ public class JdbcExecutor implements ExecutorInterface, Service {
 
     private boolean deduplicateSubflowExecution(Execution execution, ExecutorState executorState, TaskRun taskRun) {
         // There can be multiple executions for the same task, so we need to deduplicated with the worker task execution iteration
-        String deduplicationKey = taskRun.getId() + (taskRun.getIteration() == null ? "" : "-" + taskRun.getIteration());
+        String deduplicationKey = deduplicationKey(taskRun);
         State.Type current = executorState.getSubflowExecutionDeduplication().get(deduplicationKey);
 
         if (current == taskRun.getState().getCurrent()) {
-            log.trace("Duplicate SubflowExecution on execution '{}' for taskRun '{}', value '{}, taskId '{}'", execution.getId(), taskRun.getId(), taskRun.getValue(), taskRun.getTaskId());
+            log.trace("Duplicate SubflowExecution on execution '{}' for taskRun '{}', value '{}', taskId '{}', attempt '{}'", execution.getId(), taskRun.getId(), taskRun.getValue(), taskRun.getTaskId(), taskRun.getAttempts().size() + 1);
             return false;
         } else {
             executorState.getSubflowExecutionDeduplication().put(deduplicationKey, taskRun.getState().getCurrent());
@@ -929,6 +929,10 @@ public class JdbcExecutor implements ExecutorInterface, Service {
         }
     }
 
+    private String deduplicationKey(TaskRun taskRun) {
+        return taskRun.getId() + (taskRun.getAttempts() != null ? "-" + taskRun.getAttempts().size() : "") + (taskRun.getIteration() == null ? "" : "-" + taskRun.getIteration());
+    }
+
     private Executor handleFailedExecutionFromExecutor(Executor executor, Exception e) {
         Execution.FailedExecutionWithLog failedExecutionWithLog = executor.getExecution().failedExecutionFromExecutor(e);
 


Test patch:
diff --git a/core/src/test/java/io/kestra/plugin/core/flow/RetryCaseTest.java b/core/src/test/java/io/kestra/plugin/core/flow/RetryCaseTest.java
index f236a14c0..331c5b957 100644
--- a/core/src/test/java/io/kestra/plugin/core/flow/RetryCaseTest.java
+++ b/core/src/test/java/io/kestra/plugin/core/flow/RetryCaseTest.java
@@ -257,6 +257,19 @@ public class RetryCaseTest {
         assertThat(execution.getTaskRunList().get(1).attemptNumber(), is(3));
     }
 
+    public void retrySubflow() throws TimeoutException {
+        Execution execution = runnerUtils.runOne(
+            null,
+            "io.kestra.tests",
+            "retry-subflow",
+            null,
+            null
+        );
+
+        assertThat(execution.getState().getCurrent(), is(State.Type.FAILED));
+        assertThat(execution.getTaskRunList().get(0).getAttempts().size(), is(3));
+    }
+
     public void retryFlowableChild() throws TimeoutException {
         Execution execution = runnerUtils.runOne(
             null,
diff --git a/core/src/test/resources/flows/valids/retry-subflow.yaml b/core/src/test/resources/flows/valids/retry-subflow.yaml
new file mode 100644
index 000000000..0aa574996
--- /dev/null
+++ b/core/src/test/resources/flows/valids/retry-subflow.yaml
@@ -0,0 +1,11 @@
+id: retry-subflow
+namespace: io.kestra.tests
+retry:
+  type: constant
+  interval: PT1S
+  maxAttempt: 3
+tasks:
+  - id: subflow
+    type: io.kestra.plugin.core.flow.Subflow
+    namespace: io.kestra.tests
+    flowId: failed-first
\ No newline at end of file
diff --git a/jdbc/src/test/java/io/kestra/jdbc/runner/JdbcRunnerRetryTest.java b/jdbc/src/test/java/io/kestra/jdbc/runner/JdbcRunnerRetryTest.java
index 432633297..d580487c5 100644
--- a/jdbc/src/test/java/io/kestra/jdbc/runner/JdbcRunnerRetryTest.java
+++ b/jdbc/src/test/java/io/kestra/jdbc/runner/JdbcRunnerRetryTest.java
@@ -119,6 +119,11 @@ public abstract class JdbcRunnerRetryTest {
         retryCaseTest.retryFlowable();
     }
 
+    @Test
+    void retrySubflow() throws TimeoutException {
+        retryCaseTest.retrySubflow();
+    }
+
     @Test
     void retryFlowableChild() throws TimeoutException {
         retryCaseTest.retryFlowableChild();