fix(core): fix processing of execution killed with tenant (#4173)

kestra-io · Jul 5, 2024 · 5d4c4dc · 5d4c4dc
1 parent a152354
commit 5d4c4dc
Show file tree

Hide file tree

Showing 4 changed files with 21 additions and 18 deletions.
diff --git a/core/src/main/java/io/kestra/core/models/executions/ExecutionKilled.java b/core/src/main/java/io/kestra/core/models/executions/ExecutionKilled.java
@@ -9,6 +9,22 @@
 import lombok.ToString;
 import lombok.experimental.SuperBuilder;
 
+/**
+ * The Kestra event for killing an execution. A {@link ExecutionKilled} can be in two states:
+ * <p>
+ * <pre>
+ *  - {@link State#REQUESTED}: The event was requested either by an Executor or by an external request.
+ *  - {@link State#EXECUTED}: The event was consumed and processed by the Executor.
+ *  </pre>
+ *
+ *  A {@link ExecutionKilled} will always transit from {@link State#REQUESTED} to {@link State#EXECUTED}
+ *  regardless of whether the associated execution exist or not to ensure that Workers will be notified for the tasks
+ *  to be killed no matter what the circumstances.
+ *  <p>
+ *  IMPORTANT: A {@link ExecutionKilled} is considered to be a fire-and-forget event. As a result, we do not manage a
+ *  COMPLETED state, i.e., the Executor will never wait for Workers to process an executed {@link ExecutionKilled}
+ *  before considering an execution to be KILLED.
+ */
 @Getter
 @SuperBuilder
 @EqualsAndHashCode

diff --git a/core/src/main/java/io/kestra/core/models/executions/ExecutionKilledExecution.java b/core/src/main/java/io/kestra/core/models/executions/ExecutionKilledExecution.java
@@ -8,22 +8,6 @@
 import lombok.*;
 import lombok.experimental.SuperBuilder;
 
-/**
- * The Kestra event for killing an execution. A {@link ExecutionKilledExecution} can be in two states:
- * <p>
- * <pre>
- *  - {@link State#REQUESTED}: The event was requested either by an Executor or by an external request.
- *  - {@link State#EXECUTED}: The event was consumed and processed by the Executor.
- *  </pre>
- *
- *  A {@link ExecutionKilledExecution} will always transit from {@link State#REQUESTED} to {@link State#EXECUTED}
- *  regardless of whether the associated execution exist or not to ensure that Workers will be notified for the tasks
- *  to be killed no matter what the circumstances.
- *  <p>
- *  IMPORTANT: A {@link ExecutionKilledExecution} is considered to be a fire-and-forget event. As a result, we do not manage a
- *  COMPLETED state, i.e., the Executor will never wait for Workers to process an executed {@link ExecutionKilledExecution}
- *  before considering an execution to be KILLED.
- */
 @Getter
 @SuperBuilder
 @EqualsAndHashCode
@@ -47,8 +31,9 @@ public class ExecutionKilledExecution extends ExecutionKilled implements TenantI
     Boolean isOnKillCascade;
 
     public boolean isEqual(WorkerTask workerTask) {
-        return (workerTask.getTaskRun().getTenantId() == null || (workerTask.getTaskRun().getTenantId() != null && workerTask.getTaskRun().getTenantId().equals(this.tenantId))) &&
-            workerTask.getTaskRun().getExecutionId().equals(this.executionId);
+        String taskTenantId = workerTask.getTaskRun().getTenantId();
+        String taskExecutionId = workerTask.getTaskRun().getExecutionId();
+        return (taskTenantId == null || taskTenantId.equals(this.tenantId)) && taskExecutionId.equals(this.executionId);
     }
 
     @Override

diff --git a/core/src/main/java/io/kestra/core/services/ExecutionService.java b/core/src/main/java/io/kestra/core/services/ExecutionService.java
@@ -478,6 +478,7 @@ public Flux<ExecutionKilledExecution> killSubflowExecutions(final String tenantI
                 .executionId(childExecution.getId())
                 .isOnKillCascade(true)
                 .state(ExecutionKilled.State.REQUESTED) // Event will be reentrant in the Executor.
+                .tenantId(tenantId)
                 .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
@@ -706,6 +706,7 @@ private void killQueue(Either<ExecutionKilled, DeserializationException> either)
             .executionId(killedExecution.getExecutionId())
             .isOnKillCascade(false)
             .state(ExecutionKilled.State.EXECUTED)
+            .tenantId(killedExecution.getTenantId())
             .build()
         );