jprdonnelly/kestra
1
0
Fork 0
mirror of https://github.com/kestra-io/kestra.git synced 2025-12-19 18:05:41 -05:00
Code Issues Packages Projects Releases Wiki Activity

fix(jdbc,runner-memory): worker task execution has already been created if we don't wait on a flow task

Browse Source
This commit is contained in:
Loïc Mathieu
2023-10-10 12:08:52 +02:00
parent e3fb8b01a9
commit 16e54fcbe3
2 changed files with 17 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java
View File

@@ -386,13 +386,16 @@ public class JdbcExecutor implements ExecutorInterface {
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);
});

14
runner-memory/src/main/java/io/kestra/runner/memory/MemoryExecutor.java
View File

@@ -251,13 +251,17 @@ public class MemoryExecutor implements ExecutorInterface {
// 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());
}