fix(jdbc,runner-memory): worker task execution has already been creat…

…ed if we don't wait on a flow task
kestra-io · Nov 3, 2023 · 3dbbd1c · 3dbbd1c
1 parent f5e88bc
commit 3dbbd1c
Show file tree

Hide file tree

Showing 2 changed files with 17 additions and 10 deletions.
diff --git a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java
@@ -386,13 +386,16 @@ private void executionQueue(Either<Execution, DeserializationException> either)
             if (conditionService.isTerminatedWithListeners(flow, execution)) {
                 workerTaskExecutionStorage.get(execution.getId())
                     .ifPresent(workerTaskExecution -> {
-                        Flow workerTaskFlow = this.flowRepository.findByExecution(execution);
+                        // If we didn't wait for the flow execution, the worker task execution has already been created by the Executor service.
+                        if (workerTaskExecution.getTask().getWait()) {
+                            Flow workerTaskFlow = this.flowRepository.findByExecution(execution);
 
-                        WorkerTaskResult workerTaskResult = workerTaskExecution
-                            .getTask()
-                            .createWorkerTaskResult(runContextFactory, workerTaskExecution, workerTaskFlow, execution);
+                            WorkerTaskResult workerTaskResult = workerTaskExecution
+                                .getTask()
+                                .createWorkerTaskResult(runContextFactory, workerTaskExecution, workerTaskFlow, execution);
 
-                        this.workerTaskResultQueue.emit(workerTaskResult);
+                            this.workerTaskResultQueue.emit(workerTaskResult);
+                        }
 
                         workerTaskExecutionStorage.delete(workerTaskExecution);
                     });

diff --git a/runner-memory/src/main/java/io/kestra/runner/memory/MemoryExecutor.java b/runner-memory/src/main/java/io/kestra/runner/memory/MemoryExecutor.java
@@ -251,13 +251,17 @@ private void handleExecution(ExecutionState state) {
             // worker task execution
             if (conditionService.isTerminatedWithListeners(flow, execution) && WORKERTASKEXECUTIONS_WATCHER.containsKey(execution.getId())) {
                 WorkerTaskExecution workerTaskExecution = WORKERTASKEXECUTIONS_WATCHER.get(execution.getId());
-                Flow workerTaskFlow = this.flowRepository.findByExecution(execution);
 
-                WorkerTaskResult workerTaskResult = workerTaskExecution
-                    .getTask()
-                    .createWorkerTaskResult(runContextFactory, workerTaskExecution, workerTaskFlow, execution);
+                // If we didn't wait for the flow execution, the worker task execution has already been created by the Executor service.
+                if (workerTaskExecution.getTask().getWait()) {
+                    Flow workerTaskFlow = this.flowRepository.findByExecution(execution);
 
-                this.workerTaskResultQueue.emit(workerTaskResult);
+                    WorkerTaskResult workerTaskResult = workerTaskExecution
+                        .getTask()
+                        .createWorkerTaskResult(runContextFactory, workerTaskExecution, workerTaskFlow, execution);
+
+                    this.workerTaskResultQueue.emit(workerTaskResult);
+                }
 
                 WORKERTASKEXECUTIONS_WATCHER.remove(execution.getId());
             }