Repository: kestra-io/kestra (java)

Issue:
Paused execution state should count to concurrency limits on JDBC
### Explain the bug

Currently, Paused state still doesn't count to concurrency limits as it should, at least on JDBC 

## Reproducer

```yaml
id: flow-paused-concurrency
namespace: dev
concurrency:
  limit: 1
  behavior: CANCEL
tasks:
  - id: pause
    type: io.kestra.core.tasks.flows.Pause
    delay: PT30S
    tasks:
      - id: task-pause
        type: io.kestra.core.tasks.log.Log
        message: hello
```

![image](https://github.com/kestra-io/kestra/assets/86264395/0072c6dc-cbb1-4aa6-8994-b1c32a7aa8b5)


## Works with Kafka

The same works correctly with Kafka. @loicmathieu please cross-check, I was convinced we've tackled this already

![image](https://github.com/kestra-io/kestra/assets/86264395/7eeea195-d88e-4a1e-a126-5be5cd458bf9)



Pull request description:
fix(jdbc, runner-memory): paused execution didn't honor concurrency l…
…imit for FAILED and CANCELLED behavior

Fixes #2534

Patch:
diff --git a/core/src/main/java/io/kestra/core/runners/ExecutorService.java b/core/src/main/java/io/kestra/core/runners/ExecutorService.java
index cfc83659c..a5aaf7176 100644
--- a/core/src/main/java/io/kestra/core/runners/ExecutorService.java
+++ b/core/src/main/java/io/kestra/core/runners/ExecutorService.java
@@ -96,7 +96,8 @@ public class ExecutorService {
 
     public Executor process(Executor executor) {
         // previous failed (flow join can fail), just forward
-        if (!executor.canBeProcessed()) {
+        // or concurrency limit failed/cancelled the execution
+        if (!executor.canBeProcessed() || conditionService.isTerminatedWithListeners(executor.getFlow(), executor.getExecution())) {
             return executor;
         }
 
diff --git a/core/src/main/java/io/kestra/core/runners/FlowableUtils.java b/core/src/main/java/io/kestra/core/runners/FlowableUtils.java
index f5180893a..644c19419 100644
--- a/core/src/main/java/io/kestra/core/runners/FlowableUtils.java
+++ b/core/src/main/java/io/kestra/core/runners/FlowableUtils.java
@@ -277,6 +277,9 @@ public class FlowableUtils {
                 if (obj instanceof String) {
                     values.add(runContext.render((String) obj));
                 }
+                else if (obj instanceof Integer) {
+                    values.add(runContext.render(obj.toString()));
+                }
                 else if(obj instanceof Map<?, ?>) {
                     //JSON or YAML map
                     values.add(runContext.render((Map) obj));
diff --git a/ui/src/components/namespace/Editor.vue b/ui/src/components/namespace/Editor.vue
index 644281761..da8399f62 100644
--- a/ui/src/components/namespace/Editor.vue
+++ b/ui/src/components/namespace/Editor.vue
@@ -7,6 +7,7 @@
                 :value="namespace"
                 @update:model-value="namespaceUpdate"
                 allow-create
+                :is-filter="false"
             />
             <trigger-flow
                 :disabled="!flow"
@@ -41,11 +42,13 @@
     import RestoreUrl from "../../mixins/restoreUrl";
     import {apiUrl} from "override/utils/route";
     import {mapState} from "vuex";
+    import {storageKeys} from "../../utils/constants";
 
     export default {
         mixins: [RouteContext, RestoreUrl],
         methods: {
             namespaceUpdate(namespace) {
+                localStorage.setItem(storageKeys.LATEST_NAMESPACE, namespace);
                 this.$router.push({
                     params: {
                         namespace
@@ -85,7 +88,7 @@
             });
 
             // Setup namespace
-            const namespace = localStorage.getItem("defaultNamespace");
+            const namespace = localStorage.getItem(storageKeys.LATEST_NAMESPACE) ? localStorage.getItem(storageKeys.LATEST_NAMESPACE) : localStorage.getItem(storageKeys.DEFAULT_NAMESPACE);
             if (namespace) {
                 this.namespaceUpdate(namespace);
             } else if (this.namespaces?.length > 0) {
diff --git a/ui/src/components/namespace/NamespaceSelect.vue b/ui/src/components/namespace/NamespaceSelect.vue
index d090ccccb..aca07afc0 100644
--- a/ui/src/components/namespace/NamespaceSelect.vue
+++ b/ui/src/components/namespace/NamespaceSelect.vue
@@ -34,6 +34,10 @@
             allowCreate: {
                 type: Boolean,
                 default: false
+            },
+            isFilter: {
+                type: Boolean,
+                default: true
             }
         },
         emits: ["update:modelValue"],
@@ -78,7 +82,7 @@
                 });
 
                 // Remove duplicate namespaces ...
-                return _uniqBy(res,"code");
+                return _uniqBy(res,"code").filter(ns => namespaces.includes(ns.code) || this.isFilter);
             },
         }
     };
diff --git a/ui/src/utils/axios.js b/ui/src/utils/axios.js
index 0823d2394..227e84bb3 100644
--- a/ui/src/utils/axios.js
+++ b/ui/src/utils/axios.js
@@ -72,7 +72,7 @@ export default (callback, store, router) => {
         response => {
             return response
         }, errorResponse => {
-            if (errorResponse.code && (errorResponse.code === "ECONNABORTED" || errorResponse.code === "ERR_BAD_RESPONSE")) {
+            if (errorResponse?.code === "ERR_BAD_RESPONSE") {
                 store.dispatch("core/showMessage", {
                     response: errorResponse,
                     content: errorResponse,
diff --git a/ui/src/utils/constants.js b/ui/src/utils/constants.js
index d7657ef36..2d0cc18b2 100644
--- a/ui/src/utils/constants.js
+++ b/ui/src/utils/constants.js
@@ -28,7 +28,9 @@ export const storageKeys = {
     DISPLAY_EXECUTIONS_COLUMNS: "displayExecutionsColumns",
     DISPLAY_FLOW_EXECUTIONS_COLUMNS: "displayFlowExecutionsColumns",
     SELECTED_TENANT: "selectedTenant",
-    EXECUTE_FLOW_BEHAVIOUR: "executeFlowBehaviour"
+    EXECUTE_FLOW_BEHAVIOUR: "executeFlowBehaviour",
+    DEFAULT_NAMESPACE: "defaultNamespace",
+    LATEST_NAMESPACE: "latestNamespace"
 }
 
 export const executeFlowBehaviours = {


Test patch:
diff --git a/core/src/test/java/io/kestra/core/runners/FlowConcurrencyCaseTest.java b/core/src/test/java/io/kestra/core/runners/FlowConcurrencyCaseTest.java
index 4bfc9f39f..d5b4125bc 100644
--- a/core/src/test/java/io/kestra/core/runners/FlowConcurrencyCaseTest.java
+++ b/core/src/test/java/io/kestra/core/runners/FlowConcurrencyCaseTest.java
@@ -40,10 +40,8 @@ public class FlowConcurrencyCaseTest {
         assertThat(execution2.getState().getCurrent(), is(State.Type.CANCELLED));
 
         var executionResult1  = new AtomicReference<Execution>();
-        var executionResult2  = new AtomicReference<Execution>();
 
         CountDownLatch latch1 = new CountDownLatch(1);
-        CountDownLatch latch2 = new CountDownLatch(1);
 
         executionQueue.receive(e -> {
             if (e.getLeft().getId().equals(execution1.getId())) {
@@ -53,19 +51,12 @@ public class FlowConcurrencyCaseTest {
                 }
             }
 
-            if (e.getLeft().getId().equals(execution2.getId())) {
-                executionResult2.set(e.getLeft());
-                if (e.getLeft().getState().getCurrent() == State.Type.CANCELLED) {
-                    latch2.countDown();
-                }
-            }
+            // FIXME we should fail if we receive the cancel execution again but on Kafka it happens
         });
 
         latch1.await(1, TimeUnit.MINUTES);
-        latch2.await(1, TimeUnit.MINUTES);
 
         assertThat(executionResult1.get().getState().getCurrent(), is(State.Type.SUCCESS));
-        assertThat(executionResult2.get().getState().getCurrent(), is(State.Type.CANCELLED));
     }
 
     public void flowConcurrencyFail() throws TimeoutException, InterruptedException {
@@ -76,10 +67,8 @@ public class FlowConcurrencyCaseTest {
         assertThat(execution2.getState().getCurrent(), is(State.Type.FAILED));
 
         var executionResult1  = new AtomicReference<Execution>();
-        var executionResult2  = new AtomicReference<Execution>();
 
         CountDownLatch latch1 = new CountDownLatch(1);
-        CountDownLatch latch2 = new CountDownLatch(1);
 
         executionQueue.receive(e -> {
             if (e.getLeft().getId().equals(execution1.getId())) {
@@ -89,19 +78,12 @@ public class FlowConcurrencyCaseTest {
                 }
             }
 
-            if (e.getLeft().getId().equals(execution2.getId())) {
-                executionResult2.set(e.getLeft());
-                if (e.getLeft().getState().getCurrent() == State.Type.FAILED) {
-                    latch2.countDown();
-                }
-            }
+            // FIXME we should fail if we receive the cancel execution again but on Kafka it happens
         });
 
         latch1.await(1, TimeUnit.MINUTES);
-        latch2.await(1, TimeUnit.MINUTES);
 
         assertThat(executionResult1.get().getState().getCurrent(), is(State.Type.SUCCESS));
-        assertThat(executionResult2.get().getState().getCurrent(), is(State.Type.FAILED));
     }
 
     public void flowConcurrencyQueue() throws TimeoutException, InterruptedException {
@@ -199,4 +181,46 @@ public class FlowConcurrencyCaseTest {
         assertThat(executionResult2.get().getState().getHistories().get(1).getState(), is(State.Type.QUEUED));
         assertThat(executionResult2.get().getState().getHistories().get(2).getState(), is(State.Type.RUNNING));
     }
+
+    public void flowConcurrencyCancelPause() throws TimeoutException, InterruptedException {
+        Execution execution1 = runnerUtils.runOneUntilRunning(null, "io.kestra.tests", "flow-concurrency-cancel-pause", null, null, Duration.ofSeconds(30));
+        Flow flow = flowRepository
+            .findById(null, "io.kestra.tests", "flow-concurrency-cancel-pause", Optional.empty())
+            .orElseThrow();
+        Execution execution2 = runnerUtils.newExecution(flow, null, null);
+        executionQueue.emit(execution2);
+
+        assertThat(execution1.getState().isRunning(), is(true));
+        assertThat(execution2.getState().getCurrent(), is(State.Type.CREATED));
+
+        var executionResult1  = new AtomicReference<Execution>();
+        var executionResult2  = new AtomicReference<Execution>();
+
+        CountDownLatch latch1 = new CountDownLatch(1);
+        CountDownLatch latch2 = new CountDownLatch(1);
+
+        executionQueue.receive(e -> {
+            if (e.getLeft().getId().equals(execution1.getId())) {
+                executionResult1.set(e.getLeft());
+                if (e.getLeft().getState().getCurrent() == State.Type.SUCCESS) {
+                    latch1.countDown();
+                }
+            }
+
+            if (e.getLeft().getId().equals(execution2.getId())) {
+                executionResult2.set(e.getLeft());
+                if (e.getLeft().getState().getCurrent() == State.Type.CANCELLED) {
+                    latch2.countDown();
+                }
+            }
+        });
+
+        latch1.await(1, TimeUnit.MINUTES);
+        latch2.await(1, TimeUnit.MINUTES);
+
+        assertThat(executionResult1.get().getState().getCurrent(), is(State.Type.SUCCESS));
+        assertThat(executionResult2.get().getState().getCurrent(), is(State.Type.CANCELLED));
+        assertThat(executionResult2.get().getState().getHistories().get(0).getState(), is(State.Type.CREATED));
+        assertThat(executionResult2.get().getState().getHistories().get(1).getState(), is(State.Type.CANCELLED));
+    }
 }
diff --git a/core/src/test/java/io/kestra/core/tasks/flows/EachParallelTest.java b/core/src/test/java/io/kestra/core/tasks/flows/EachParallelTest.java
index 30d1590a4..3ab4b0a53 100644
--- a/core/src/test/java/io/kestra/core/tasks/flows/EachParallelTest.java
+++ b/core/src/test/java/io/kestra/core/tasks/flows/EachParallelTest.java
@@ -1,6 +1,7 @@
 package io.kestra.core.tasks.flows;
 
 import io.kestra.core.models.flows.State;
+import io.kestra.core.queues.QueueException;
 import org.junit.jupiter.api.Test;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.runners.AbstractMemoryRunnerTest;
@@ -28,4 +29,10 @@ public class EachParallelTest extends AbstractMemoryRunnerTest {
         assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
         assertThat(execution.getTaskRunList(), hasSize(11));
     }
+
+    @Test
+    void parallelInteger() throws TimeoutException, QueueException {
+        Execution execution = runnerUtils.runOne(null, "io.kestra.tests", "each-parallel-Integer");
+        assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
+    }
 }
diff --git a/core/src/test/resources/flows/valids/each-parallel-Integer.yml b/core/src/test/resources/flows/valids/each-parallel-Integer.yml
new file mode 100644
index 000000000..0837b5068
--- /dev/null
+++ b/core/src/test/resources/flows/valids/each-parallel-Integer.yml
@@ -0,0 +1,11 @@
+id: each-parallel-Integer
+namespace: io.kestra.tests
+
+tasks:
+  - id: parallel
+    type: io.kestra.core.tasks.flows.EachParallel
+    value: [1, 2, 3] # this works ["1", "2", "3"]
+    tasks:
+      - id: return
+        type: io.kestra.core.tasks.debugs.Return
+        format: "{{taskrun.value}}"
\ No newline at end of file
diff --git a/core/src/test/resources/flows/valids/flow-concurrency-cancel-pause.yml b/core/src/test/resources/flows/valids/flow-concurrency-cancel-pause.yml
new file mode 100644
index 000000000..5c58d8e06
--- /dev/null
+++ b/core/src/test/resources/flows/valids/flow-concurrency-cancel-pause.yml
@@ -0,0 +1,15 @@
+id: flow-concurrency-cancel-pause
+namespace: io.kestra.tests
+
+concurrency:
+  behavior: CANCEL
+  limit: 1
+
+tasks:
+  - id: pause
+    type: io.kestra.core.tasks.flows.Pause
+    delay: PT1S
+    tasks:
+      - id: post-pause
+        type: io.kestra.core.tasks.log.Log
+        message: Post-pause
\ No newline at end of file
diff --git a/jdbc/src/test/java/io/kestra/jdbc/runner/JdbcRunnerTest.java b/jdbc/src/test/java/io/kestra/jdbc/runner/JdbcRunnerTest.java
index 42c733d57..f768c5165 100644
--- a/jdbc/src/test/java/io/kestra/jdbc/runner/JdbcRunnerTest.java
+++ b/jdbc/src/test/java/io/kestra/jdbc/runner/JdbcRunnerTest.java
@@ -302,4 +302,9 @@ public abstract class JdbcRunnerTest {
     void concurrencyQueuePause() throws TimeoutException, InterruptedException  {
         flowConcurrencyCaseTest.flowConcurrencyQueuePause();
     }
+
+    @Test
+    void concurrencyCancelPause() throws TimeoutException, InterruptedException  {
+        flowConcurrencyCaseTest.flowConcurrencyCancelPause();
+    }
 }