Repository: kestra-io/kestra
Issue: ForEachItem -> concurrency.limit bug
Labels: bug

### Describe the issue

Using the ForEachItem flow with a few hundred entries, we needed to use the concurrency limit. We had a limit of 4 for testing. The first 4 executions take this limit into account, but as soon as a first execution is completed, all the others go into running quickly. On a few executions, this is not a problem but on several hundred or even thousands Kestra crashes.

### Environment

- Kestra Version: 0.13.3 & 0.14.4

Issue comments:

Pull request review comments:

Patch that resolved the issue:
diff --git a/core/src/main/java/io/kestra/core/runners/ExecutorState.java b/core/src/main/java/io/kestra/core/runners/ExecutorState.java
index b448c976da..1588dbd36d 100644
--- a/core/src/main/java/io/kestra/core/runners/ExecutorState.java
+++ b/core/src/main/java/io/kestra/core/runners/ExecutorState.java
@@ -14,7 +14,6 @@ public class ExecutorState {
     private Map<String, State.Type> workerTaskDeduplication = new ConcurrentHashMap<>();
     private Map<String, String> childDeduplication = new ConcurrentHashMap<>();
     private Map<String, State.Type> subflowExecutionDeduplication = new ConcurrentHashMap<>();
-    private Boolean flowTriggerDeduplication = false;
 
     public ExecutorState(String executionId) {
         this.executionId = executionId;
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 6dadc959ce..d6ce5baca1 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java
@@ -482,37 +482,6 @@ private void executionQueue(Either<Execution, DeserializationException> either)
                     });
             }
 
-            // multiple condition
-            if (
-                conditionService.isTerminatedWithListeners(flow, execution) &&
-                    this.deduplicateFlowTrigger(execution, executorState)
-            ) {
-                flowTriggerService.computeExecutionsFromFlowTriggers(execution, allFlows, Optional.of(multipleConditionStorage))
-                    .forEach(this.executionQueue::emit);
-            }
-
-            // when terminated: handle queued executions and subflow executions
-            if (conditionService.isTerminatedWithListeners(flow, execution)) {
-                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() != null && subflowExecution.getParentTask().waitForExecution()) {
-                            sendSubflowExecutionResult(execution, subflowExecution, subflowExecution.getParentTaskRun().withState(execution.getState().getCurrent()));
-                        }
-
-                        subflowExecutionStorage.delete(subflowExecution);
-                    });
-
-                if (flow.getConcurrency() != null && flow.getConcurrency().getBehavior() == Concurrency.Behavior.QUEUE) {
-                    // as the execution is terminated, we can check if there exist a queued execution and submit it to the execution queue
-                    executionQueuedStorage.pop(flow.getTenantId(),
-                        flow.getNamespace(),
-                        flow.getId(),
-                        queued -> executionQueue.emit(queued.withState(State.Type.RUNNING))
-                    );
-                }
-            }
-
             return Pair.of(
                 executor,
                 executorState
@@ -793,6 +762,35 @@ private void toExecution(Executor executor) {
         if (executorService.canBePurged(executor)) {
             executorStateStorage.delete(executor.getExecution());
         }
+
+        // handle actions on terminated state
+        // the terminated state can only come from the execution queue, and in this case we always have a flow in the executor
+        if (executor.getFlow() != null && conditionService.isTerminatedWithListeners(executor.getFlow(), executor.getExecution())) {
+            Execution execution = executor.getExecution();
+            // handle flow triggers
+            flowTriggerService.computeExecutionsFromFlowTriggers(execution, allFlows, Optional.of(multipleConditionStorage))
+                .forEach(this.executionQueue::emit);
+
+            // purge subflow execution storage
+            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() != null && subflowExecution.getParentTask().waitForExecution()) {
+                        sendSubflowExecutionResult(execution, subflowExecution, subflowExecution.getParentTaskRun().withState(execution.getState().getCurrent()));
+                    }
+
+                    subflowExecutionStorage.delete(subflowExecution);
+                });
+
+            // check if there exist a queued execution and submit it to the execution queue
+            if (executor.getFlow().getConcurrency() != null && executor.getFlow().getConcurrency().getBehavior() == Concurrency.Behavior.QUEUE) {
+                executionQueuedStorage.pop(executor.getFlow().getTenantId(),
+                    executor.getFlow().getNamespace(),
+                    executor.getFlow().getId(),
+                    queued -> executionQueue.emit(queued.withState(State.Type.RUNNING))
+                );
+            }
+        }
     }
 
     private Flow transform(Flow flow, Execution execution) {
@@ -890,18 +888,6 @@ private boolean deduplicateSubflowExecution(Execution execution, ExecutorState e
         }
     }
 
-    private boolean deduplicateFlowTrigger(Execution execution, ExecutorState executorState) {
-        Boolean flowTriggerDeduplication = executorState.getFlowTriggerDeduplication();
-
-        if (flowTriggerDeduplication) {
-            log.trace("Duplicate Flow Trigger on execution '{}'", execution.getId());
-            return false;
-        } else {
-            executorState.setFlowTriggerDeduplication(true);
-            return true;
-        }
-    }
-
     private Executor handleFailedExecutionFromExecutor(Executor executor, Exception e) {
         Execution.FailedExecutionWithLog failedExecutionWithLog = executor.getExecution().failedExecutionFromExecutor(e);
 


Test patch: