Repository: kestra-io/kestra
File: jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java (Java)

Code before:
                    Map<String, Object> outputs = (Map<String, Object>) execution.getTrigger().getVariables().get("taskRunOutputs");
                    Variables variables = variablesService.of(StorageContext.forExecution(executor.getExecution()), outputs);
                    SubflowExecutionEnd subflowExecutionEnd = new SubflowExecutionEnd(executor.getExecution(), parentExecutionId, taskRunId, taskId, execution.getState().getCurrent(), variables);
                    this.subflowExecutionEndQueue.emit(subflowExecutionEnd);
                }

                // purge SLA monitors
                if (!ListUtils.isEmpty(executor.getFlow().getSla()) && executor.getFlow().getSla().stream().anyMatch(ExecutionMonitoringSLA.class::isInstance)) {
                    slaMonitorStorage.purge(executor.getExecution().getId());
                }

                // check if there exist a queued execution and submit it to the execution queue
                if (executor.getFlow().getConcurrency() != null) {

                    // decrement execution concurrency limit
                    // if an execution was queued but never running, it would have never been counted inside the concurrency limit and should not lead to popping a new queued execution
                    // this could only happen for KILLED execution.
                    boolean queuedThenKilled = execution.getState().getCurrent() == State.Type.KILLED
                        && execution.getState().getHistories().stream().anyMatch(h -> h.getState().isQueued())
                        && execution.getState().getHistories().stream().noneMatch(h -> h.getState().onlyRunning());
                    boolean concurrencyShortCircuitState = Concurrency.possibleTransitions(execution.getState().getCurrent())
                        && execution.getState().getHistories().get(execution.getState().getHistories().size() - 2).getState().isCreated();
                    if (!queuedThenKilled && !concurrencyShortCircuitState) {
                        var currentFlow = executor.getFlow();

                        if (!(execution.getState().getCurrent() == State.Type.KILLED && executor.getFrom().contains("addWorkerTaskResult"))) {
                            concurrencyLimitStorage.decrement(executor.getFlow());

                            int currentConcurencyLimit = this.concurrencyLimitStorage.fetchOne(currentFlow).getRunning();

                            if (executor.getFlow().getConcurrency().getBehavior() == Concurrency.Behavior.QUEUE
                                && currentConcurencyLimit < currentFlow.getConcurrency().getLimit()) {
                                var finalFlow = executor.getFlow();
                                executionQueuedStorage.pop(executor.getFlow().getTenantId(),
                                    executor.getFlow().getNamespace(),
                                    executor.getFlow().getId(),
                                    throwBiConsumer((dslContext, queued) -> {
                                        var newExecution = queued.withState(State.Type.RUNNING);
                                        concurrencyLimitStorage.increment(dslContext, finalFlow);
                                        executionQueue.emit(newExecution);
                                        metricRegistry.counter(MetricRegistry.METRIC_EXECUTOR_EXECUTION_POPPED_COUNT, MetricRegistry.METRIC_EXECUTOR_EXECUTION_POPPED_COUNT_DESCRIPTION, metricRegistry.tags(newExecution)).increment();

                                        // process flow triggers to allow listening on RUNNING state after a QUEUED state
                                        processFlowTriggers(newExecution);
                                    })
                                );
                            }
                        }
                    }
                }


Review comment (bug):
I'm not fan of using the `from` here, we never use this anywhere like that, it has been added only for debugging purpose I would not rely on it.
The purpose of the boolean just before are to do the same kind of check but based on the states which would be better

Code after:

            if (log.isDebugEnabled()) {
                executorService.log(log, false, executor);
            }

            // the terminated state can come from the execution queue, in this case we always have a flow in the executor
            // or from a worker task in an afterExecution block, in this case we need to load the flow
            if (executor.getFlow() == null && executor.getExecution().getState().isTerminated()) {
                executor = executor.withFlow(findFlowOrThrow(executor.getExecution()));
            }
            boolean isTerminated = executor.getFlow() != null && executionService.isTerminated(executor.getFlow(), executor.getExecution());

            // purge the executionQueue
            // IMPORTANT: this must be done before emitting the last execution message so that all consumers are notified that the execution ends.
            // NOTE: we may also purge ExecutionKilled events, but as there may not be a lot of them, it may not be worth it.
            if (cleanExecutionQueue && isTerminated) {
                ((JdbcQueue<Execution>) executionQueue).deleteByKey(executor.getExecution().getId());
            }

            // emit for other consumers than the executor if no failure
            if (hasFailure) {
                this.executionQueue.emit(executor.getExecution());
            } else {
                ((JdbcQueue<Execution>) this.executionQueue).emitOnly(null, executor.getExecution());
            }

            Execution execution = executor.getExecution();
            // handle flow triggers on state change
            if (!execution.getState().getCurrent().equals(executor.getOriginalState())) {
                processFlowTriggers(execution);
            }

            // handle actions on terminated state
            if (isTerminated) {
                // if there is a parent, we send a subflow execution result to it
                if (ExecutableUtils.isSubflow(execution)) {
                    // locate the parent execution to find the parent task run
                    String parentExecutionId = (String) execution.getTrigger().getVariables().get("executionId");
                    String taskRunId = (String) execution.getTrigger().getVariables().get("taskRunId");
                    String taskId = (String) execution.getTrigger().getVariables().get("taskId");
                    @SuppressWarnings("unchecked")
                    Map<String, Object> outputs = (Map<String, Object>) execution.getTrigger().getVariables().get("taskRunOutputs");
                    Variables variables = variablesService.of(StorageContext.forExecution(executor.getExecution()), outputs);
                    SubflowExecutionEnd subflowExecutionEnd = new SubflowExecutionEnd(executor.getExecution(), parentExecutionId, taskRunId, taskId, execution.getState().getCurrent(), variables);
                    this.subflowExecutionEndQueue.emit(subflowExecutionEnd);
                }

                // purge SLA monitors
                if (!ListUtils.isEmpty(executor.getFlow().getSla()) && executor.getFlow().getSla().stream().anyMatch(ExecutionMonitoringSLA.class::isInstance)) {
                    slaMonitorStorage.purge(executor.getExecution().getId());
                }