Repository: kestra-io/kestra (java)

Issue:
Unqueue an execution to a different state than RUNNING
### Feature description

Starting with 0.20, we can unqueue a queued execution (due to concurrency limit exceeded), unqueue will set the execution to running and plays nicely with the concurrency limit behavior.

But in some case, you just don't want to run these executions to completion, you may want for example to cancel them as they are queued for too long and no longer accurate.

The best would be, when we unqueue an execution, to be able to choose the state of the execution, not all states makes sense so maybe RUNNING (default), CANCELLED or FAILED.

Pull request description:
feat(executions): Allow unqueuing to states other than RUNNING #5939
<!-- Thanks for submitting a Pull Request to Kestra. To help us review your contribution, please follow the guidelines below:

- Make sure that your commits follow the [conventional commits](https://www.conventionalcommits.org/en/v1.0.0/) specification e.g. `feat(ui): add a new navigation menu item` or `fix(core): fix a bug in the core model` or `docs: update the README.md`. This will help us automatically generate the changelog.
- The title should briefly summarize the proposed changes.
- Provide a short overview of the change and the value it adds.
- Share a flow example to help the reviewer understand and QA the change.
- Use "closes" to automatically close an issue. For example, `closes #1234` will close issue #1234. -->

### What changes are being made and why?
Allows queued executions to be moved to RUNNING (default), CANCELLED, or FAILED states when unqueued.

closes #5939 


Patch:
diff --git a/core/src/main/java/io/kestra/core/runners/ExecutionRunning.java b/core/src/main/java/io/kestra/core/runners/ExecutionRunning.java
index 76282a423..1c2a6f702 100644
--- a/core/src/main/java/io/kestra/core/runners/ExecutionRunning.java
+++ b/core/src/main/java/io/kestra/core/runners/ExecutionRunning.java
@@ -30,5 +30,5 @@ public class ExecutionRunning {
         return IdUtils.fromPartsAndSeparator('|', this.tenantId, this.namespace, this.flowId, this.execution.getId());
     }
 
-    public enum ConcurrencyState { CREATED, RUNNING, QUEUED }
+    public enum ConcurrencyState { CREATED, RUNNING, QUEUED, CANCELLED, FAILED }
 }
diff --git a/core/src/main/java/io/kestra/core/services/ConcurrencyLimitService.java b/core/src/main/java/io/kestra/core/services/ConcurrencyLimitService.java
index 98b8b45a4..00f15e9b6 100644
--- a/core/src/main/java/io/kestra/core/services/ConcurrencyLimitService.java
+++ b/core/src/main/java/io/kestra/core/services/ConcurrencyLimitService.java
@@ -5,18 +5,32 @@ import io.kestra.core.models.flows.State;
 import io.kestra.core.queues.QueueException;
 import jakarta.inject.Singleton;
 
+import java.util.EnumSet;
+import java.util.Set;
+
 @Singleton
 public class ConcurrencyLimitService {
 
+    protected static final Set<State.Type> VALID_TARGET_STATES =
+        EnumSet.of(State.Type.RUNNING, State.Type.CANCELLED, State.Type.FAILED);
+
     /**
      * Unqueue a queued execution.
      *
      * @throws IllegalArgumentException in case the execution is not queued.
      */
-    public Execution unqueue(Execution execution) throws QueueException {
+    public Execution unqueue(Execution execution, State.Type state) throws QueueException {
         if (execution.getState().getCurrent() != State.Type.QUEUED) {
             throw new IllegalArgumentException("Only QUEUED execution can be unqueued");
         }
-        return execution.withState(State.Type.RUNNING);
+
+        state = (state == null) ? State.Type.RUNNING : state;
+
+        // Validate the target state, throwing an exception if the state is invalid
+        if (!VALID_TARGET_STATES.contains(state)) {
+            throw new IllegalArgumentException("Invalid target state: " + state + ". Valid states are: " + VALID_TARGET_STATES);
+        }
+
+        return execution.withState(state);
     }
 }
diff --git a/core/src/main/java/io/kestra/core/services/ExecutionService.java b/core/src/main/java/io/kestra/core/services/ExecutionService.java
index 24d5fc9b1..692770ed0 100644
--- a/core/src/main/java/io/kestra/core/services/ExecutionService.java
+++ b/core/src/main/java/io/kestra/core/services/ExecutionService.java
@@ -843,7 +843,7 @@ public class ExecutionService {
         }
 
         if (execution.getState().getCurrent() == State.Type.QUEUED) {
-            return concurrencyLimitService.unqueue(execution);
+            return concurrencyLimitService.unqueue(execution,State.Type.RUNNING);
         }
 
         if (execution.getState().getCurrent() == State.Type.PAUSED) {
diff --git a/jdbc/src/main/java/io/kestra/jdbc/services/JdbcConcurrencyLimitService.java b/jdbc/src/main/java/io/kestra/jdbc/services/JdbcConcurrencyLimitService.java
index 95c8510cf..0651ee3a2 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/services/JdbcConcurrencyLimitService.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/services/JdbcConcurrencyLimitService.java
@@ -19,13 +19,20 @@ public class JdbcConcurrencyLimitService extends ConcurrencyLimitService {
     private AbstractJdbcExecutionQueuedStorage storage;
 
     @Override
-    public Execution unqueue(Execution execution) throws QueueException {
+    public Execution unqueue(Execution execution, State.Type state) throws QueueException {
         if (execution.getState().getCurrent() != State.Type.QUEUED) {
             throw new IllegalArgumentException("Only QUEUED execution can be unqueued");
         }
 
         storage.remove(execution);
 
-        return execution.withState(State.Type.RUNNING);
+        state = (state == null) ? State.Type.RUNNING : state;
+
+        // Validate the target state, throwing an exception if the state is invalid
+        if (!VALID_TARGET_STATES.contains(state)) {
+            throw new IllegalArgumentException("Invalid target state: " + state + ". Valid states are: " + VALID_TARGET_STATES);
+        }
+
+        return execution.withState(state);
     }
 }
diff --git a/ui/src/components/executions/Executions.vue b/ui/src/components/executions/Executions.vue
index fc8d2bea2..c4d3ca54e 100644
--- a/ui/src/components/executions/Executions.vue
+++ b/ui/src/components/executions/Executions.vue
@@ -115,7 +115,7 @@
                                         <el-dropdown-item v-if="canUpdate" :icon="PauseBox" @click="pauseExecutions()">
                                             {{ $t("pause") }}
                                         </el-dropdown-item>
-                                        <el-dropdown-item v-if="canUpdate" :icon="QueueFirstInLastOut" @click="unqueueExecutions()">
+                                        <el-dropdown-item v-if="canUpdate" :icon="QueueFirstInLastOut" @click="unqueueDialogVisible = true">
                                             {{ $t("unqueue") }}
                                         </el-dropdown-item>
                                         <el-dropdown-item v-if="canUpdate" :icon="RunFast" @click="forceRunExecutions()">
@@ -362,6 +362,45 @@
         </template>
     </el-dialog>
 
+    <el-dialog v-if="unqueueDialogVisible" v-model="unqueueDialogVisible" destroy-on-close :append-to-body="true">
+        <template #header>
+            <h5>{{ $t("confirmation") }}</h5>
+        </template>
+
+        <template #default>
+            <p v-html="$t('unqueue title multiple', {count: queryBulkAction ? total : selection.length})" />
+
+            <el-select
+                :required="true"
+                v-model="selectedStatus"
+                :persistent="false"
+            >
+                <el-option
+                    v-for="item in unQueuestates"
+                    :key="item.code"
+                    :value="item.code"
+                >
+                    <template #default>
+                        <status size="small" :label="false" class="me-1" :status="item.code" />
+                        <span v-html="item.label" />
+                    </template>
+                </el-option>
+            </el-select>
+        </template>
+
+        <template #footer>
+            <el-button @click="unqueueDialogVisible = false">
+                {{ $t('cancel') }}
+            </el-button>
+            <el-button
+                type="primary"
+                @click="unqueueExecutions()"
+            >
+                {{ $t('ok') }}
+            </el-button>
+        </template>
+    </el-dialog>
+
     <el-dialog v-if="isOpenReplayModal" v-model="isOpenReplayModal" :id="Utils.uid()" destroy-on-close :append-to-body="true" align-center>
         <template #header>
             <h5>{{ $t("confirmation") }}</h5>
@@ -572,6 +611,7 @@
                 lastRefreshDate: new Date(),
                 isOpenReplayModal: false,
                 changeStatusDialogVisible: false,
+                unqueueDialogVisible: false,
                 selectedStatus: undefined,
                 loading: false
             };
@@ -648,6 +688,17 @@
                     };
                 });
             },
+            unQueuestates() {
+                return [State.RUNNING, State.CANCELLED, State.FAILED].map(value => ({
+                    code: value,
+                    label: this.$t("unqueue as", {status: value}),
+                }));
+            },
+            executionsCount() {
+                return [...this.daily].reduce((a, b) => {
+                    return a + Object.values(b.executionCounts).reduce((a, b) => a + b, 0);
+                }, 0);
+            },
             selectedNamespace(){
                 return this.namespace !== null && this.namespace !== undefined ? this.namespace : this.$route.query?.namespace;
             },
@@ -820,8 +871,10 @@
                 );
             },
             unqueueExecutions() {
-                this.genericConfirmAction(
-                    "bulk unqueue",
+                this.unqueueDialogVisible = false;
+                this.actionOptions.newStatus = this.selectedStatus;
+
+                this.genericConfirmCallback(
                     "execution/queryUnqueueExecution",
                     "execution/bulkUnqueueExecution",
                     "executions unqueue"
diff --git a/ui/src/components/executions/Overview.vue b/ui/src/components/executions/Overview.vue
index db2809421..6ffeacab1 100644
--- a/ui/src/components/executions/Overview.vue
+++ b/ui/src/components/executions/Overview.vue
@@ -14,6 +14,9 @@
                             <chevron-up v-if="isExpanded" />
                             <chevron-down v-else />
                         </span>
+                        <span v-if="!errorLogs">
+                            {{ $t('error detected') }}
+                        </span>
                     </div>
                 </template>
                 <div v-if="isExpanded && errorLogs" class="error-stack">
diff --git a/ui/src/components/executions/Unqueue.vue b/ui/src/components/executions/Unqueue.vue
index c5f46450b..28fee419d 100644
--- a/ui/src/components/executions/Unqueue.vue
+++ b/ui/src/components/executions/Unqueue.vue
@@ -2,7 +2,7 @@
     <component
         :is="component"
         :icon="QueueFirstInLastOut"
-        @click="click"
+        @click="isDrawerOpen = !isDrawerOpen"
         v-if="enabled"
         class="ms-0 me-1"
     >
@@ -11,8 +11,31 @@
 
     <el-dialog v-if="isDrawerOpen" v-model="isDrawerOpen" destroy-on-close :append-to-body="true">
         <template #header>
-            <span v-html="$t('unqueue title', {id: execution.id})" />
+            <span v-html="$t('unqueue')" />
         </template>
+
+        <template #default>
+            <p v-html="$t('unqueue title', {id: execution.id})" />
+
+            <el-select
+                :required="true"
+                v-model="selectedStatus"
+                :persistent="false"
+            >
+                <el-option
+                    v-for="item in states"
+                    :key="item.code"
+                    :value="item.code"
+                    :disabled="item.disabled"
+                >
+                    <template #default>
+                        <status size="small" :label="true" class="me-1" :status="item.code" />
+                        <span v-html="item.label" />
+                    </template>
+                </el-option>
+            </el-select>           
+        </template>
+
         <template #footer>
             <el-button :icon="QueueFirstInLastOut" type="primary" @click="unqueue()" native-type="submit">
                 {{ $t('unqueue') }}
@@ -30,8 +53,10 @@
     import permission from "../../models/permission";
     import action from "../../models/action";
     import {State} from "@kestra-io/ui-libs"
+    import Status from "../../components/Status.vue";
 
     export default {
+        components: {Status},
         props: {
             execution: {
                 type: Object,
@@ -45,19 +70,15 @@
         data() {
             return {
                 isDrawerOpen: false,
+                selectedStatus: State.RUNNING,
             };
         },
         methods: {
-            click() {
-                this.$toast()
-                    .confirm(this.$t("unqueue confirm", {id: this.execution.id}), () => {
-                        return this.unqueue();
-                    });
-            },
             unqueue() {
                 this.$store
                     .dispatch("execution/unqueue", {
-                        id: this.execution.id
+                        id: this.execution.id,
+                        state: this.selectedStatus
                     })
                     .then(() => {
                         this.isDrawerOpen = false;
@@ -68,6 +89,12 @@
         computed: {
             ...mapState("auth", ["user"]),
             ...mapState("execution", ["flow"]),
+            states() {
+                return [State.RUNNING, State.CANCELLED, State.FAILED].map(value => ({                
+                    code: value,
+                    label: this.$t("unqueue as", {status: value}),                
+                }));
+            },
             enabled() {
                 if (!(this.user && this.user.isAllowed(permission.EXECUTION, action.UPDATE, this.execution.namespace))) {
                     return false;
diff --git a/ui/src/stores/executions.js b/ui/src/stores/executions.js
index 9042dcb57..e397d1955 100644
--- a/ui/src/stores/executions.js
+++ b/ui/src/stores/executions.js
@@ -299,17 +299,17 @@ export default {
             return this.$http.post(`${apiUrl(this)}/executions/labels/by-ids`,  options)
         },
         unqueue(_, options) {
-            return this.$http.post(`${apiUrl(this)}/executions/${options.id}/unqueue`);
+            return this.$http.post(`${apiUrl(this)}/executions/${options.id}/unqueue?state=${options.state}`);
         },
         bulkUnqueueExecution(_, options) {
             return this.$http.post(
-                `${apiUrl(this)}/executions/unqueue/by-ids`,
+                `${apiUrl(this)}/executions/unqueue/by-ids?state=${options.newStatus}`,
                 options.executionsId
             )
         },
         queryUnqueueExecution(_, options) {
             return this.$http.post(
-                `${apiUrl(this)}/executions/unqueue/by-query`,
+                `${apiUrl(this)}/executions/unqueue/by-query?state=${options.newStatus}`,
                 {},
                 {params: options}
             )
diff --git a/ui/src/translations/en.json b/ui/src/translations/en.json
index 4ecbecdf9..13935ea54 100644
--- a/ui/src/translations/en.json
+++ b/ui/src/translations/en.json
@@ -195,6 +195,7 @@
     "change state": "Change state",
     "change state done": "Task state has been updated",
     "change state confirm": "Are you sure you want to change the task run state for the <code>{task}</code> task in execution <code>{id}</code>?<br/>It will then restart the execution from the next task and restart the parent execution if any.",
+    "change queue confirm": "Are you sure you want to change the queue state for the execution <code>{id}</code>?<br/>",
     "change state current state": "Current status is:",
     "change state hint": {
       "WARNING": [
@@ -221,6 +222,7 @@
     "change execution state confirm": "Are you sure you want to change the state of the execution <code>{id}</code>?",
     "change execution state done": "Execution state updated",
     "mark as": "Mark as <code>{status}</code>",
+    "unqueue as": "Unqueue as <code>{status}</code>",
     "kill": "Kill",
     "kill parents and subflow": "Kill parents and subflows",
     "kill only parents": "Kill parents only",
@@ -235,7 +237,8 @@
     "pause confirm": "Are you sure you want to pause execution <code>{id}</code>?",
     "pause done": "Execution is paused",
     "unqueue": "Unqueue",
-    "unqueue title": "Unqueue execution <code>{id}</code>.<br/>Note that it will not obey the flow concurrency limit, so more executions could be running than the allowed limit.",
+    "unqueue title": "Are you sure you want to Unqueue execution <code>{id}</code>?<br/><br/>Note that it will not obey the flow concurrency limit, so more executions could be running than the allowed limit.",
+    "unqueue title multiple": "Are you sure you want to Unqueue <code>{count}</code> executions?<br/><br/>Note that it will not obey the flow concurrency limit, so more executions could be running than the allowed limit.",
     "unqueue confirm": "Are you sure you want to unqueue execution <code>{id}</code>?",
     "unqueue done": "Execution is unqueued",
     "force run": "Force run",
diff --git a/webserver/src/main/java/io/kestra/webserver/controllers/api/ExecutionController.java b/webserver/src/main/java/io/kestra/webserver/controllers/api/ExecutionController.java
index bcf5c435c..8889c6387 100644
--- a/webserver/src/main/java/io/kestra/webserver/controllers/api/ExecutionController.java
+++ b/webserver/src/main/java/io/kestra/webserver/controllers/api/ExecutionController.java
@@ -1920,14 +1920,15 @@ public class ExecutionController {
     @Post(uri = "/{executionId}/unqueue")
     @Operation(tags = {"Executions"}, summary = "Unqueue an execution")
     public Execution unqueueExecution(
-        @Parameter(description = "The execution id") @PathVariable String executionId
+        @Parameter(description = "The execution id") @PathVariable String executionId,
+        @Parameter(description = "The new state of the execution") @Nullable @QueryValue State.Type state
     ) throws Exception {
         Optional<Execution> execution = executionRepository.findById(tenantService.resolveTenant(), executionId);
         if (execution.isEmpty()) {
             return null;
         }
 
-        Execution restart = concurrencyLimitService.unqueue(execution.get());
+        Execution restart = concurrencyLimitService.unqueue(execution.get(), state);
         executionQueue.emit(restart);
         eventPublisher.publishEvent(new CrudEvent<>(restart, execution.get(), CrudEventType.UPDATE));
 
@@ -1940,7 +1941,8 @@ public class ExecutionController {
     @ApiResponse(responseCode = "200", description = "On success", content = {@Content(schema = @Schema(implementation = BulkResponse.class))})
     @ApiResponse(responseCode = "422", description = "Unqueued with errors", content = {@Content(schema = @Schema(implementation = BulkErrorResponse.class))})
     public MutableHttpResponse<?> unqueueExecutionsByIds(
-        @RequestBody(description = "The list of executions id") @Body List<String> executionsId
+        @RequestBody(description = "The list of executions id") @Body List<String> executionsId,
+        @Parameter(description = "The new state of the unqueued executions") @Nullable @QueryValue State.Type state
     ) throws Exception {
         List<Execution> executions = new ArrayList<>();
         Set<ManualConstraintViolation<String>> invalids = new HashSet<>();
@@ -1977,7 +1979,7 @@ public class ExecutionController {
             );
         }
         for (Execution execution : executions) {
-            Execution restart = concurrencyLimitService.unqueue(execution);
+            Execution restart = concurrencyLimitService.unqueue(execution, state);
             executionQueue.emit(restart);
             eventPublisher.publishEvent(new CrudEvent<>(restart, execution, CrudEventType.UPDATE));
         }
@@ -2004,7 +2006,8 @@ public class ExecutionController {
         @Deprecated @Parameter(description = "A state filter") @Nullable @QueryValue List<State.Type> state,
         @Deprecated @Parameter(description = "A labels filter as a list of 'key:value'") @Nullable @QueryValue @Format("MULTI") List<String> labels,
         @Deprecated @Parameter(description = "The trigger execution id") @Nullable @QueryValue String triggerExecutionId,
-        @Deprecated @Parameter(description = "A execution child filter") @Nullable @QueryValue ExecutionRepositoryInterface.ChildFilter childFilter
+        @Deprecated @Parameter(description = "A execution child filter") @Nullable @QueryValue ExecutionRepositoryInterface.ChildFilter childFilter,
+        @Parameter(description = "The new state of the unqueued executions") @Nullable @QueryValue State.Type newState
     ) throws Exception {
         filters = RequestUtils.getFiltersOrDefaultToLegacyMapping(
             filters,
@@ -2026,7 +2029,7 @@ public class ExecutionController {
 
         var ids = getExecutionIds(filters);
 
-        return unqueueExecutionsByIds(ids);
+        return unqueueExecutionsByIds(ids, newState);
     }
 
     @ExecuteOn(TaskExecutors.IO)


Test patch:
diff --git a/webserver/src/test/java/io/kestra/webserver/controllers/api/ExecutionControllerRunnerTest.java b/webserver/src/test/java/io/kestra/webserver/controllers/api/ExecutionControllerRunnerTest.java
index 0b51eb116..fd9cc982c 100644
--- a/webserver/src/test/java/io/kestra/webserver/controllers/api/ExecutionControllerRunnerTest.java
+++ b/webserver/src/test/java/io/kestra/webserver/controllers/api/ExecutionControllerRunnerTest.java
@@ -1359,6 +1359,24 @@ class ExecutionControllerRunnerTest {
         assertThat(notRunning.getStatus().getCode()).isEqualTo(HttpStatus.UNPROCESSABLE_ENTITY.getCode());
     }
 
+    @Test
+    @LoadFlows({"flows/valids/flow-concurrency-queue.yml",
+        "flows/valids/minimal.yaml"})
+    void shouldUnqueueAQueuedFlowToCancelledState() throws QueueException, TimeoutException {
+        // run a first flow so the second is queued
+        runnerUtils.runOneUntilRunning(TENANT_ID, "io.kestra.tests", "flow-concurrency-queue");
+        Execution result1 = runUntilQueued("io.kestra.tests", "flow-concurrency-queue");
+
+        var cancelResponse = client.toBlocking().exchange(
+            HttpRequest.POST("/api/v1/executions/" + result1.getId() + "/unqueue?state=CANCELLED", null)
+        );
+        assertThat(cancelResponse.getStatus().getCode()).isEqualTo(HttpStatus.OK.getCode());
+
+        Optional<Execution> cancelledExecution = executionRepositoryInterface.findById(TENANT_ID, result1.getId());
+        assertThat(cancelledExecution.isPresent()).isTrue();
+        assertThat(cancelledExecution.get().getState().getCurrent()).isEqualTo(State.Type.CANCELLED);
+    }
+
     @Test
     @LoadFlows({"flows/valids/flow-concurrency-queue.yml"})
     void shouldUnqueueExecutionByIdsQueuedFlows() throws TimeoutException, QueueException {