Repository: kestra-io/kestra (java)

Issue:
Investiguate improvement to JDBC queues cleaning
### Issue description

Today, the JDBC `queues` table is cleaned by the JDBCCleaner which, periodically (1h by default) cleans all records that have an `updated` date that is more than 7 days (by default) old.

In case of high volume of executions (or moderate volume if complex orchestration flows) this table can grow up pretty quickly in the order of millions or rows.

There is multiple things here:
- The cleaning is done only on the `updated` column, this means that we cannot clean too aggressively (not hours) as they may be multiple consumer of a single queue row.
- The cleaning is done by default after 7 days which is a lot, we advise some user to lower it to 1d which seems to didn't cause any issue.
- We may be able to purge immediately after consumption as there is no usage of a message in the queue after consumption;

Pull request description:
feat(jdbc): purge execution queue early
Purge eagerly queue messages from the executor: execution, worker job, worker task result.

Other messages will be purged by the JdbcCleaner. 

Fixes #5055 

Patch:
diff --git a/.github/workflows/workflow-build-artifacts.yml b/.github/workflows/workflow-build-artifacts.yml
index fef7ce10d..e1f14dca2 100644
--- a/.github/workflows/workflow-build-artifacts.yml
+++ b/.github/workflows/workflow-build-artifacts.yml
@@ -10,13 +10,13 @@ on:
         type: string
     outputs:
       docker-tag:
-        value: ${{ steps.vars.outputs.tag }}
+        value: ${{ jobs.build.outputs.tag }}
         description: "The Docker image Tag for Kestra"
       docker-artifact-name:
-        value: ${{ steps.vars.outputs.artifact }}
+        value: ${{ jobs.build.outputs.artifact }}
         description: "The GitHub artifact containing the Kestra docker image name."
       plugins:
-        value: ${{ steps.plugins-list.outputs.plugins }}
+        value: ${{ jobs.build.outputs.plugins }}
         description: "The Kestra plugins list used for the build."
 
 jobs:
diff --git a/.github/workflows/workflow-release.yml b/.github/workflows/workflow-release.yml
index 0fba53000..517a45336 100644
--- a/.github/workflows/workflow-release.yml
+++ b/.github/workflows/workflow-release.yml
@@ -6,7 +6,7 @@ on:
       plugin-version:
         description: "Kestra version"
         default: 'LATEST'
-        required: true
+        required: false
         type: string
     secrets:
       DOCKERHUB_USERNAME:
@@ -49,11 +49,10 @@ jobs:
     secrets:
         DOCKERHUB_USERNAME: ${{ secrets.DOCKERHUB_USERNAME }}
         DOCKERHUB_PASSWORD: ${{ secrets.DOCKERHUB_PASSWORD }}
-        GOOGLE_SERVICE_ACCOUNT: ${{ secrets.GOOGLE_SERVICE_ACCOUNT }}
 
   Maven:
     name: Publish Maven
-    uses: ./.github/workflows/workflow-build-artifacts.yml
+    uses: ./.github/workflows/workflow-publish-maven.yml
     secrets:
       SONATYPE_USER: ${{ secrets.SONATYPE_USER }}
       SONATYPE_PASSWORD: ${{ secrets.SONATYPE_PASSWORD }}
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 9f5721818..e9f03d8de 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java
@@ -81,7 +81,7 @@ public class JdbcExecutor implements ExecutorInterface, Service {
 
     @Inject
     @Named(QueueFactoryInterface.WORKERJOB_NAMED)
-    private QueueInterface<WorkerJob> workerTaskQueue;
+    private QueueInterface<WorkerJob> workerJobQueue;
 
     @Inject
     @Named(QueueFactoryInterface.WORKERTASKRESULT_NAMED)
@@ -368,7 +368,7 @@ public class JdbcExecutor implements ExecutorInterface, Service {
                         workerJobRunningRepository.deleteByKey(workerTaskRunning.uid());
                     } else {
                         try {
-                            workerTaskQueue.emit(WorkerTask.builder()
+                            workerJobQueue.emit(WorkerTask.builder()
                                 .taskRun(workerTaskRunning.getTaskRun().onRunningResend())
                                 .task(workerTaskRunning.getTask())
                                 .runContext(workerTaskRunning.getRunContext())
@@ -395,7 +395,7 @@ public class JdbcExecutor implements ExecutorInterface, Service {
                 // WorkerTriggerRunning
                 if (workerJobRunning instanceof WorkerTriggerRunning workerTriggerRunning) {
                     try {
-                        workerTaskQueue.emit(WorkerTrigger.builder()
+                        workerJobQueue.emit(WorkerTrigger.builder()
                             .trigger(workerTriggerRunning.getTrigger())
                             .conditionContext(workerTriggerRunning.getConditionContext())
                             .triggerContext(workerTriggerRunning.getTriggerContext())
@@ -533,7 +533,7 @@ public class JdbcExecutor implements ExecutorInterface, Service {
                                             workerTaskResultQueue.emit(new WorkerTaskResult(workerTask.getTaskRun().withState(State.Type.SKIPPED)));
                                         } else {
                                             if (workerTask.getTask().isSendToWorkerTask()) {
-                                                workerTaskQueue.emit(workerGroupService.resolveGroupFromJob(workerTask).map(WorkerGroup::getKey).orElse(null), workerTask);
+                                                workerJobQueue.emit(workerGroupService.resolveGroupFromJob(workerTask).map(group -> group.getKey()).orElse(null), workerTask);
                                             }
                                             if (workerTask.getTask().isFlowable()) {
                                                 workerTaskResultQueue.emit(new WorkerTaskResult(workerTask.getTaskRun().withState(State.Type.RUNNING)));
@@ -940,14 +940,24 @@ public class JdbcExecutor implements ExecutorInterface, Service {
                 executorService.log(log, false, executor);
             }
 
-            // emit for other consumer than executor if no failure
+            // the terminated state can only come from the execution queue, in this case we always have a flow in the executor
+            boolean isTerminated = executor.getFlow() != null && conditionService.isTerminatedWithListeners(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 (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());
             }
 
-            // delete if ended
+            // delete the execution from the state storage if ended
             if (executorService.canBePurged(executor)) {
                 executorStateStorage.delete(executor.getExecution());
             }
@@ -960,8 +970,7 @@ public class JdbcExecutor implements ExecutorInterface, Service {
             }
 
             // 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())) {
+            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
@@ -987,6 +996,18 @@ public class JdbcExecutor implements ExecutorInterface, Service {
                         throwConsumer(queued -> executionQueue.emit(queued.withState(State.Type.RUNNING)))
                     );
                 }
+
+                // Purge the workerTaskResultQueue and the workerJobQueue
+                // IMPORTANT: this is safe as only the executor is listening to WorkerTaskResult,
+                // and we are sure at this stage that all WorkerJob has been listened and processed by the Worker.
+                // If any of these assumptions changed, this code would not be safe anymore.
+                if (!ListUtils.isEmpty(executor.getExecution().getTaskRunList())) {
+                    List<String> taskRunKeys = executor.getExecution().getTaskRunList().stream()
+                        .map(taskRun -> taskRun.getId())
+                        .toList();
+                    ((JdbcQueue<WorkerTaskResult>) workerTaskResultQueue).deleteByKeys(taskRunKeys);
+                    ((JdbcQueue<WorkerJob>) workerJobQueue).deleteByKeys(taskRunKeys);
+                }
             }
         } catch (QueueException e) {
             if (!ignoreFailure) {
diff --git a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcQueue.java b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcQueue.java
index 99175242a..3193bf186 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcQueue.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcQueue.java
@@ -3,6 +3,7 @@ package io.kestra.jdbc.runner;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.base.CaseFormat;
+import com.google.common.collect.Iterables;
 import io.kestra.core.exceptions.DeserializationException;
 import io.kestra.core.metrics.MetricRegistry;
 import io.kestra.core.models.executions.Execution;
@@ -163,6 +164,41 @@ public abstract class JdbcQueue<T> implements QueueInterface<T> {
         // and the queue has its own cleaner, which we better not mess with, as the 'queues' table is selected with a lock.
     }
 
+    /**
+     * Delete all messages of the queue for this key.
+     * This is used to purge a queue for a specific key.
+     */
+    public void deleteByKey(String key) throws QueueException {
+        dslContextWrapper.transaction(configuration -> {
+            int deleted = DSL
+                .using(configuration)
+                .delete(this.table)
+                .where(buildTypeCondition(this.cls.getName()))
+                .and(AbstractJdbcRepository.field("key").eq(key))
+                .execute();
+            log.debug("Cleaned {} records for key {}", deleted, key);
+        });
+    }
+
+    /**
+     * Delete all messages of the queue for a set of keys.
+     * This is used to purge a queue for specific keys.
+     */
+    public void deleteByKeys(List<String> keys) throws QueueException {
+        // process in batches of 100 items to avoid too big IN clause
+        Iterables.partition(keys, 100).forEach(batch -> {
+            dslContextWrapper.transaction(configuration -> {
+                int deleted = DSL
+                    .using(configuration)
+                    .delete(this.table)
+                    .where(buildTypeCondition(this.cls.getName()))
+                    .and(AbstractJdbcRepository.field("key").in(batch))
+                    .execute();
+                log.debug("Cleaned {} records for keys {}", deleted, batch);
+            });
+        });
+    }
+
     protected Result<Record> receiveFetch(DSLContext ctx, String consumerGroup, Integer offset) {
         return this.receiveFetch(ctx, consumerGroup, offset, true);
     }


Test patch:
diff --git a/.github/workflows/workflow-test.yml b/.github/workflows/workflow-test.yml
index 30a82f810..a96e9b2c1 100644
--- a/.github/workflows/workflow-test.yml
+++ b/.github/workflows/workflow-test.yml
@@ -29,7 +29,6 @@ jobs:
 
   backend:
     name: Backend - Tests
-    if: "needs.file-changes.outputs.backend == 'true'"
     uses: ./.github/workflows/workflow-backend-test.yml
     secrets:
       GITHUB_AUTH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
diff --git a/core/src/test/resources/sanity-checks/request.yaml b/core/src/test/resources/sanity-checks/request.yaml
new file mode 100644
index 000000000..78fa84de2
--- /dev/null
+++ b/core/src/test/resources/sanity-checks/request.yaml
@@ -0,0 +1,16 @@
+id: request
+namespace: sanitycheck.plugin.core.http
+
+tasks:
+  - id: request
+    type: io.kestra.plugin.core.http.Request
+    uri: https://www.google.com
+    method: GET
+    options:
+      connectionPoolIdleTimeout: PT1M
+
+  - id: assert
+    type: io.kestra.plugin.core.execution.Assert
+    errorMessage: "Invalid response code {{ outputs.request.code }}"
+    conditions:
+      - "{{ outputs.request.code == 200 }}"
\ No newline at end of file