Repository: kestra-io/kestra
Issue: A WorkerTaskExecution with an Invalid Task can fail the executor
Labels: bug
quick-win

### Explain the bug

A WorkerTaskExecution with an Invalid Task can fail the executor

```
2023-10-13 15:29:13,028 ERROR jdbc-queue_4 .c.u.ThreadUncaughtExceptionHandlers Caught an exception in Thread[jdbc-queue_4,5,main]. Shutting down.
io.kestra.core.exceptions.DeserializationException: com.fasterxml.jackson.databind.exc.InvalidTypeIdException: Could not resolve type id 'io.kestra.core.tasks.flows.ForEachItem$SubFlowTask' as a subtype of `io.kestra.core.tasks.flows.Flow`: no such class found
 at [Source: (String)"{"task": {"id": "each-subflow", "type": "io.kestra.core.tasks.flows.ForEachItem$SubFlowTask", "wait": true, "flowId": "per-item", "inputs": {"items": "{{ taskrun.items }}"}, "namespace": "dev", "transmitFailed": true}, "taskRun": {"id": "4EMuw9AF7RjtG5pyLPJ2jk", "items": "kestra:///dev/each-item/executions/Kd09l4qTI0Ngpl6U78LKk/tasks/each/Y3eVpVwevRwNyVDuKmCRK/bach-2654.ion", "state": {"current": "CREATED", "duration": 1.230850996, "histories": [{"date": "2023-10-11T15:14:03.358Z", "state": "CRE"[truncated 902 chars]; line: 1, column: 41] (through reference chain: io.kestra.core.runners.WorkerTaskExecution["task"])
	at io.kestra.jdbc.AbstractJdbcRepository.deserialize(AbstractJdbcRepository.java:167)
	at io.kestra.jdbc.AbstractJdbcRepository.map(AbstractJdbcRepository.java:118)
	at java.base/java.util.Optional.map(Optional.java:260)
	at io.kestra.jdbc.AbstractJdbcRepository.fetchOne(AbstractJdbcRepository.java:173)
	at io.kestra.jdbc.runner.AbstractJdbcWorkerTaskExecutionStorage.lambda$get$0(AbstractJdbcWorkerTaskExecutionStorage.java:34)
	at org.jooq.impl.DefaultDSLContext.lambda$transactionResult0$3(DefaultDSLContext.java:552)
	at org.jooq.impl.Tools$3$1.block(Tools.java:5888)
	at java.base/java.util.concurrent.ForkJoinPool.unmanagedBlock(ForkJoinPool.java:3463)
	at java.base/java.util.concurrent.ForkJoinPool.managedBlock(ForkJoinPool.java:3434)
	at org.jooq.impl.Tools$3.get(Tools.java:5885)
	at org.jooq.impl.DefaultDSLContext.transactionResult0(DefaultDSLContext.java:600)
	at org.jooq.impl.DefaultDSLContext.transactionResult(DefaultDSLContext.java:524)
	at io.kestra.jdbc.JooqDSLContextWrapper.lambda$transactionResult$2(JooqDSLContextWrapper.java:67)
	at net.jodah.failsafe.Functions.lambda$get$0(Functions.java:48)
	at net.jodah.failsafe.RetryPolicyExecutor.lambda$supply$0(RetryPolicyExecutor.java:66)
	at net.jodah.failsafe.FallbackExecutor.lambda$supply$0(FallbackExecutor.java:45)
	at net.jodah.failsafe.Execution.executeSync(Execution.java:128)
	at net.jodah.failsafe.FailsafeExecutor.call(FailsafeExecutor.java:379)
	at net.jodah.failsafe.FailsafeExecutor.get(FailsafeExecutor.java:68)
	at io.kestra.core.utils.RetryUtils$Instance.wrap(RetryUtils.java:134)
	at io.kestra.core.utils.RetryUtils$Instance.runRetryIf(RetryUtils.java:93)
	at io.kestra.jdbc.JooqDSLContextWrapper.transactionResult(JooqDSLContextWrapper.java:65)
	at io.kestra.jdbc.runner.AbstractJdbcWorkerTaskExecutionStorage.get(AbstractJdbcWorkerTaskExecutionStorage.java:25)
	at io.kestra.jdbc.runner.JdbcExecutor.lambda$executionQueue$17(JdbcExecutor.java:387)
	at io.kestra.jdbc.repository.AbstractJdbcExecutionRepository.lambda$lock$21(AbstractJdbcExecutionRepository.java:697)
	at org.jooq.impl.DefaultDSLContext.lambda$transactionResult0$3(DefaultDSLContext.java:552)
	at org.jooq.impl.Tools$3$1.block(Tools.java:5888)
	at java.base/java.util.concurrent.ForkJoinPool.unmanagedBlock(ForkJoinPool.java:3463)
	at java.base/java.util.concurrent.ForkJoinPool.managedBlock(ForkJoinPool.java:3434)
	at org.jooq.impl.Tools$3.get(Tools.java:5885)
	at org.jooq.impl.DefaultDSLContext.transactionResult0(DefaultDSLContext.java:600)
	at org.jooq.impl.DefaultDSLContext.transactionResult(DefaultDSLContext.java:524)
	at io.kestra.jdbc.JooqDSLContextWrapper.lambda$transactionResult$2(JooqDSLContextWrapper.java:67)
	at net.jodah.failsafe.Functions.lambda$get$0(Functions.java:48)
	at net.jodah.failsafe.RetryPolicyExecutor.lambda$supply$0(RetryPolicyExecutor.java:66)
	at net.jodah.failsafe.FallbackExecutor.lambda$supply$0(FallbackExecutor.java:45)
	at net.jodah.failsafe.Execution.executeSync(Execution.java:128)
	at net.jodah.failsafe.FailsafeExecutor.call(FailsafeExecutor.java:379)
	at net.jodah.failsafe.FailsafeExecutor.get(FailsafeExecutor.java:68)
	at io.kestra.core.utils.RetryUtils$Instance.wrap(RetryUtils.java:134)
	at io.kestra.core.utils.RetryUtils$Instance.runRetryIf(RetryUtils.java:93)
	at io.kestra.jdbc.JooqDSLContextWrapper.transactionResult(JooqDSLContextWrapper.java:65)
	at io.kestra.jdbc.repository.AbstractJdbcExecutionRepository.lock(AbstractJdbcExecutionRepository.java:679)
	at io.kestra.jdbc.runner.JdbcExecutor.executionQueue(JdbcExecutor.java:285)
	at java.base/java.util.ArrayList.forEach(ArrayList.java:1511)
	at io.kestra.jdbc.runner.JdbcQueue.lambda$receive$6(JdbcQueue.java:191)
	at io.kestra.jdbc.runner.JdbcQueue.lambda$receiveImpl$9(JdbcQueue.java:237)
	at io.kestra.jdbc.runner.JdbcQueue.lambda$poll$10(JdbcQueue.java:260)
	at io.micrometer.core.instrument.internal.TimedRunnable.run(TimedRunnable.java:49)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
	at java.base/java.lang.Thread.run(Thread.java:833)
```

### Environment Information

- Kestra Version: 0.12
- Operating System and Java Version (if not using Kestra Docker image):


Issue comments:

Pull request review comments:

Patch that resolved the issue:
diff --git a/jdbc/src/main/java/io/kestra/jdbc/runner/AbstractJdbcSubflowExecutionStorage.java b/jdbc/src/main/java/io/kestra/jdbc/runner/AbstractJdbcSubflowExecutionStorage.java
index 3ebc3ca7f9..64dccd180a 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/runner/AbstractJdbcSubflowExecutionStorage.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/runner/AbstractJdbcSubflowExecutionStorage.java
@@ -1,6 +1,11 @@
 package io.kestra.jdbc.runner;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.kestra.core.exceptions.DeserializationException;
+import io.kestra.core.models.executions.Execution;
+import io.kestra.core.models.executions.TaskRun;
 import io.kestra.core.runners.SubflowExecution;
+import io.kestra.core.serializers.JacksonMapper;
 import io.kestra.jdbc.repository.AbstractJdbcRepository;
 import org.jooq.DSLContext;
 import org.jooq.Field;
@@ -13,6 +18,7 @@
 import java.util.Optional;
 
 public abstract class AbstractJdbcSubflowExecutionStorage extends AbstractJdbcRepository {
+    private final static ObjectMapper MAPPER = JacksonMapper.ofJson();
     protected io.kestra.jdbc.AbstractJdbcRepository<SubflowExecution<?>> jdbcRepository;
 
     public AbstractJdbcSubflowExecutionStorage(io.kestra.jdbc.AbstractJdbcRepository jdbcRepository) {
@@ -31,7 +37,18 @@ public Optional<SubflowExecution<?>> get(String executionId) {
                         AbstractJdbcRepository.field("key").eq(executionId)
                     );
 
-                return this.jdbcRepository.fetchOne(select);
+                try {
+                    return this.jdbcRepository.fetchOne(select);
+                } catch (DeserializationException deserializationException) {
+                    // we may fail to deserialize a SubflowExecution if we fail to deserialize its task
+                    var jsonNode = MAPPER.readTree(deserializationException.getRecord());
+                    var taskRun = MAPPER.treeToValue(jsonNode.get("parentTaskRun"), TaskRun.class);
+                    var execution = MAPPER.treeToValue(jsonNode.get("execution"), Execution.class);
+                    return Optional.of(SubflowExecution.builder()
+                        .parentTaskRun(taskRun)
+                        .execution(execution)
+                        .build());
+                }
             });
     }
 
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 1a302efafd..31d52a91a5 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java
@@ -37,7 +37,6 @@
 import io.kestra.jdbc.repository.AbstractJdbcWorkerJobRunningRepository;
 import io.micronaut.context.ApplicationContext;
 import io.micronaut.context.annotation.Value;
-import io.micronaut.http.HttpResponse;
 import io.micronaut.transaction.exceptions.CannotCreateTransactionException;
 import jakarta.inject.Inject;
 import jakarta.inject.Named;
@@ -49,7 +48,6 @@
 
 import java.io.IOException;
 import java.time.Duration;
-import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.*;
@@ -468,7 +466,7 @@ private void executionQueue(Either<Execution, DeserializationException> either)
                 subflowExecutionStorage.get(execution.getId())
                     .ifPresent(subflowExecution -> {
                         // If we didn't wait for the flow execution, the worker task execution has already been created by the Executor service.
-                        if (subflowExecution.getParentTask().waitForExecution()) {
+                        if (subflowExecution.getParentTask() != null && subflowExecution.getParentTask().waitForExecution()) {
                             sendSubflowExecutionResult(execution, subflowExecution, subflowExecution.getParentTaskRun().withState(execution.getState().getCurrent()));
                         }
 


Test patch:
diff --git a/core/src/test/java/io/kestra/core/runners/DeserializationIssuesCaseTest.java b/core/src/test/java/io/kestra/core/runners/DeserializationIssuesCaseTest.java
index 8da685845d..2aeb63f51d 100644
--- a/core/src/test/java/io/kestra/core/runners/DeserializationIssuesCaseTest.java
+++ b/core/src/test/java/io/kestra/core/runners/DeserializationIssuesCaseTest.java
@@ -149,6 +149,67 @@ public class DeserializationIssuesCaseTest {
         }
         """;
 
+    public static final String INVALID_SUBFLOW_EXECUTION_KEY = "1XKpihp8y2m3KEHR0hVEKN";
+    public static final String INVALID_SUBFLOW_EXECUTION_VALUE =  """
+    {
+      "execution": {
+        "id": "1XKpihp8y2m3KEHR0hVEKN",
+        "state": {
+          "current": "CREATED",
+          "duration": 0.000201173,
+          "histories": [
+            {
+              "date": "2024-01-10T13:48:32.752Z",
+              "state": "CREATED"
+            }
+          ],
+          "startDate": "2024-01-10T13:48:32.752Z"
+        },
+        "flowId": "hello-world",
+        "deleted": false,
+        "trigger": {
+          "id": "subflow",
+          "type": "io.kestra.notfound.Invalid",
+          "variables": {
+            "flowId": "subflox",
+            "namespace": "company.team",
+            "executionId": "4NzSyOQBYj1CxVg3bTghbZ",
+            "flowRevision": 1
+          }
+        },
+        "namespace": "company.team",
+        "originalId": "1XKpihp8y2m3KEHR0hVEKN",
+        "flowRevision": 2
+      },
+      "parentTask": {
+        "id": "subflow",
+        "type": "io.kestra.notfound.Invalid"
+      },
+      "parentTaskRun": {
+        "id": "6Gc6Dkk7medsWtg1WJfZpN",
+        "state": {
+          "current": "RUNNING",
+          "duration": 0.039446974,
+          "histories": [
+            {
+              "date": "2024-01-10T13:48:32.713Z",
+              "state": "CREATED"
+            },
+            {
+              "date": "2024-01-10T13:48:32.752Z",
+              "state": "RUNNING"
+            }
+          ],
+          "startDate": "2024-01-10T13:48:32.713Z"
+        },
+        "flowId": "subflox",
+        "taskId": "subflow",
+        "namespace": "company.team",
+        "executionId": "4NzSyOQBYj1CxVg3bTghbZ"
+      }
+    }
+    """;
+
     @Inject
     @Named(QueueFactoryInterface.WORKERTASKRESULT_NAMED)
     protected QueueInterface<WorkerTaskResult> workerTaskResultQueue;
diff --git a/jdbc-postgres/src/test/java/io/kestra/runner/postgres/PostgresSubflowExecutionStorageTest.java b/jdbc-postgres/src/test/java/io/kestra/runner/postgres/PostgresSubflowExecutionStorageTest.java
index ef8ae3e41c..9dde7a46f4 100644
--- a/jdbc-postgres/src/test/java/io/kestra/runner/postgres/PostgresSubflowExecutionStorageTest.java
+++ b/jdbc-postgres/src/test/java/io/kestra/runner/postgres/PostgresSubflowExecutionStorageTest.java
@@ -1,7 +1,17 @@
 package io.kestra.runner.postgres;
 
+import io.kestra.core.runners.DeserializationIssuesCaseTest;
 import io.kestra.jdbc.runner.AbstractSubflowExecutionTest;
+import org.jooq.Field;
+import org.jooq.JSONB;
+import org.jooq.impl.DSL;
 
-class PostgresSubflowExecutionStorageTest extends AbstractSubflowExecutionTest {
+import java.util.Map;
 
+class PostgresSubflowExecutionStorageTest extends AbstractSubflowExecutionTest {
+    @Override
+    protected Map<Field<Object>, Object> persistFields() {
+        return Map.of(io.kestra.jdbc.repository.AbstractJdbcRepository.field("value"),
+            DSL.val(JSONB.valueOf(DeserializationIssuesCaseTest.INVALID_SUBFLOW_EXECUTION_VALUE)));
+    }
 }
\ No newline at end of file
diff --git a/jdbc/src/test/java/io/kestra/jdbc/runner/AbstractSubflowExecutionTest.java b/jdbc/src/test/java/io/kestra/jdbc/runner/AbstractSubflowExecutionTest.java
index 27b11976ec..066af5ee73 100644
--- a/jdbc/src/test/java/io/kestra/jdbc/runner/AbstractSubflowExecutionTest.java
+++ b/jdbc/src/test/java/io/kestra/jdbc/runner/AbstractSubflowExecutionTest.java
@@ -2,16 +2,20 @@
 
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.executions.TaskRun;
+import io.kestra.core.runners.DeserializationIssuesCaseTest;
 import io.kestra.core.runners.SubflowExecution;
 import io.kestra.core.tasks.flows.Subflow;
 import io.kestra.core.utils.IdUtils;
 import io.kestra.jdbc.JdbcTestUtils;
 import io.micronaut.test.extensions.junit5.annotation.MicronautTest;
 import jakarta.inject.Inject;
+import org.jooq.Field;
+import org.jooq.impl.DSL;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 
 import static org.hamcrest.MatcherAssert.assertThat;
@@ -48,6 +52,25 @@ void suite() throws Exception {
         assertThat(find.isPresent(), is(false));
     }
 
+    @Test
+    void deserializationIssue() {
+        // insert an invalid subflowExecution
+        var subflowExecution = SubflowExecution.builder()
+            .execution(Execution.builder().id(DeserializationIssuesCaseTest.INVALID_SUBFLOW_EXECUTION_KEY).build())
+            .build();
+        Map<Field<Object>, Object> fields = persistFields();
+        subflowExecutionStorage.jdbcRepository.persist(subflowExecution, fields);
+
+        // load it
+        Optional<SubflowExecution<?>> find = subflowExecutionStorage.get(DeserializationIssuesCaseTest.INVALID_SUBFLOW_EXECUTION_KEY);
+        assertThat(find.isPresent(), is(true));
+    }
+
+    protected Map<Field<Object>, Object>  persistFields() {
+        return Map.of(io.kestra.jdbc.repository.AbstractJdbcRepository.field("value"),
+            DeserializationIssuesCaseTest.INVALID_SUBFLOW_EXECUTION_VALUE);
+    }
+
     @BeforeEach
     protected void init() {
         jdbcTestUtils.drop();