Repository: kestra-io/kestra
Issue: Paused execution state should count to concurrency limits on JDBC
Labels: bug

### 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)



Issue comments:
Will double check as I fixed this for JDBC last week.
I can reproduce the issue, pause count as running ... but cancel state is not correctly handled with pause task.
I'll have a look.

Pull request review comments:

Patch that resolved the issue:
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 cfc83659c6..a5aaf7176c 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 Executor checkConcurrencyLimit(Executor executor, Flow flow, Execution ex
 
     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;
         }
 


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 4bfc9f39f1..d5b4125bca 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 void flowConcurrencyCancel() throws TimeoutException, InterruptedExceptio
         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 void flowConcurrencyCancel() throws TimeoutException, InterruptedExceptio
                 }
             }
 
-            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 void flowConcurrencyFail() throws TimeoutException, InterruptedException
         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 void flowConcurrencyFail() throws TimeoutException, InterruptedException
                 }
             }
 
-            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 void flowConcurrencyQueuePause() throws TimeoutException, InterruptedExce
         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/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 0000000000..5c58d8e06e
--- /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 42c733d57e..f768c51659 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 @@ void concurrencyQueue() throws TimeoutException, InterruptedException  {
     void concurrencyQueuePause() throws TimeoutException, InterruptedException  {
         flowConcurrencyCaseTest.flowConcurrencyQueuePause();
     }
+
+    @Test
+    void concurrencyCancelPause() throws TimeoutException, InterruptedException  {
+        flowConcurrencyCaseTest.flowConcurrencyCancelPause();
+    }
 }