Repository: kestra-io/kestra (java)

Issue:
If statement condition evaluation fail
### Describe the issue

Example flow:
```yaml
id: for_loop_example
namespace: company.team

tasks:
  - id: for_each
    type: io.kestra.plugin.core.flow.ForEach
    values: ["value 1", "value 2", "value 3"]
    tasks:
      - id: before_if
        type: io.kestra.plugin.core.debug.Return
        format: "Before if: {{ taskrun.value }}"
      - id: if
        type: io.kestra.plugin.core.flow.If
        condition: "{{ taskrun.value equals 'value 2' }}"
        then:
          - id: after_if
            type: io.kestra.plugin.core.debug.Return
            format: "After if: {{ parent.taskrun.value }}"
```
This works on `latest` 19.8. But on `develop`, it throws the following error when it gets to the `if` task.

```
Unable to resolve state from the Flowable task: Unable to find the 'evaluationResult' output, this may indicate that the condition evaluation fail, check your execution logs for more information.
```

 I tried replacing the condition with just true to remove any issues with the taskrun expression but the same error. I was able to get an if statement working without ForEach. Example used below:

<details>
<summary>Example that worked with an if statement</summary>

```yaml
id: if
namespace: company.team

inputs:
  - id: string
    type: STRING
    required: true

tasks:
  - id: if
    type: io.kestra.plugin.core.flow.If
    condition: "{{ inputs.string == 'Condition' }}"
    then:
      - id: when_true
        type: io.kestra.plugin.core.log.Log
        message: "Condition was true"
    else:
      - id: when_false
        type: io.kestra.plugin.core.log.Log
        message: "Condition was false"
```

</details>

### Environment

- Kestra Version: develop
- Commit Hash: 4259466
- Edition: OSS

Pull request description:
fix(core): If in a flowable task
Fixes #5812

Patch:
diff --git a/core/src/main/java/io/kestra/core/models/tasks/WorkerGroup.java b/core/src/main/java/io/kestra/core/models/tasks/WorkerGroup.java
index f4ac91bc2..4b5ed6da8 100644
--- a/core/src/main/java/io/kestra/core/models/tasks/WorkerGroup.java
+++ b/core/src/main/java/io/kestra/core/models/tasks/WorkerGroup.java
@@ -16,4 +16,12 @@ import jakarta.validation.constraints.Pattern;
 public class WorkerGroup {
     @Pattern(regexp="^[a-zA-Z0-9][a-zA-Z0-9_-]*")
     private String key;
+
+    private Fallback fallback;
+
+    public enum Fallback {
+        FAIL,
+        WAIT,
+        CANCEL,
+    }
 }
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 04968302c..f0a25e0a8 100644
--- a/core/src/main/java/io/kestra/core/runners/ExecutorService.java
+++ b/core/src/main/java/io/kestra/core/runners/ExecutorService.java
@@ -22,6 +22,7 @@ import io.kestra.core.models.tasks.retrys.AbstractRetry;
 import io.kestra.core.services.ConditionService;
 import io.kestra.core.services.ExecutionService;
 import io.kestra.core.services.LogService;
+import io.kestra.core.services.WorkerGroupService;
 import io.kestra.core.utils.ListUtils;
 import io.kestra.plugin.core.flow.Pause;
 import io.kestra.plugin.core.flow.Subflow;
@@ -75,6 +76,9 @@ public class ExecutorService {
     @Inject
     private ExecutionService executionService;
 
+    @Inject
+    private WorkerGroupService workerGroupService;
+
     protected FlowExecutorInterface flowExecutorInterface() {
         // bean is injected late, so we need to wait
         if (this.flowExecutorInterface == null) {
@@ -437,7 +441,6 @@ public class ExecutorService {
         }
 
         return executor.withTaskRun(
-            // TODO - saveFlowableOutput seems to be only useful for Template
             this.saveFlowableOutput(nextTaskRuns, executor),
             "handleNext"
         );
@@ -743,29 +746,47 @@ public class ExecutorService {
                         .task(task)
                         .build();
                     // Get worker group
-                    String workerGroup = Optional.ofNullable(workerTask.getTask())
-                        .map(Task::getWorkerGroup)
-                        .map(WorkerGroup::getKey)
-                        .orElse(null);
-                    // Check if the worker group exist
-                    String tenantId = executor.getFlow().getTenantId();
-                    if (workerGroupExecutorInterface.isWorkerGroupExistForKey(workerGroup, tenantId)) {
-                        // Check whether at-least one worker is available
-                        if (workerGroupExecutorInterface.isWorkerGroupAvailableForKey(workerGroup)) {
-                            return workerTask;
+                    Optional<WorkerGroup> workerGroup = workerGroupService.resolveGroupFromJob(workerTask);
+                    if (workerGroup.isPresent()) {
+                        // Check if the worker group exist
+                        String tenantId = executor.getFlow().getTenantId();
+                        String workerGroupKey = workerGroup.get().getKey();
+                        if (workerGroupExecutorInterface.isWorkerGroupExistForKey(workerGroupKey, tenantId)) {
+                            // Check whether at-least one worker is available
+                            if (workerGroupExecutorInterface.isWorkerGroupAvailableForKey(workerGroupKey)) {
+                                return workerTask;
+                            } else {
+                                WorkerGroup.Fallback fallback = workerGroup.map(wg -> wg.getFallback()).orElse(WorkerGroup.Fallback.WAIT);
+                                return switch(fallback) {
+                                    case FAIL -> {
+                                        runContext.logger()
+                                            .error("No workers are available for worker group '{}', failing the task.", workerGroupKey);
+                                        yield workerTask.withTaskRun(workerTask.getTaskRun().fail());
+                                    }
+                                    case CANCEL -> {
+                                        runContext.logger()
+                                            .info("No workers are available for worker group '{}', canceling the task.", workerGroupKey);
+                                        yield workerTask.withTaskRun(workerTask.getTaskRun().withState(State.Type.CANCELLED));
+                                    }
+                                    case WAIT -> {
+                                        runContext.logger()
+                                            .info("No workers are available for worker group '{}', waiting for one to be available.", workerGroupKey);
+                                        yield workerTask;
+                                    }
+                                };
+                            }
                         } else {
                             runContext.logger()
-                                .error("Cannot run task. No workers are available for worker group '" + workerGroup + "'.");
+                                .error("Cannot run task. No worker group exist for key '{}'.", workerGroupKey);
+                            // fail the task-run because no worker can run the task
+                            return workerTask.withTaskRun(workerTask.getTaskRun().fail());
                         }
                     } else {
-                        runContext.logger()
-                            .error("Cannot run task. No worker group exist for key '" + workerGroup + "'.");
+                        return workerTask;
                     }
-                    // fail the task-run because no worker can run the task
-                    return workerTask.withTaskRun(workerTask.getTaskRun().fail());
                 })
             )
-            .collect(Collectors.groupingBy(workerTask -> workerTask.getTaskRun().getState().isFailed()));
+            .collect(Collectors.groupingBy(workerTask -> workerTask.getTaskRun().getState().isFailed() || workerTask.getTaskRun().getState().getCurrent() == State.Type.CANCELLED));
 
         if (workerTasks.isEmpty()) {
             return executor;
@@ -773,21 +794,20 @@ public class ExecutorService {
 
         Executor executorToReturn = executor;
 
-        // Handle WorkerTasks for FAILED TaskRun
-        List<WorkerTask> workerTasksFailed = workerTasks.get(true);
-        if (workerTasksFailed != null) {
-            List<WorkerTaskResult> failed = workerTasksFailed
+        // Ends FAILED or CANCELLED task runs by creating worker task results
+        List<WorkerTask> endedTasks = workerTasks.get(true);
+        if (endedTasks != null && !endedTasks.isEmpty()) {
+            List<WorkerTaskResult> failed = endedTasks
                 .stream()
-                .filter(workerTask -> workerTask.getTaskRun().getState().isFailed())
                 .map(workerTask -> WorkerTaskResult.builder().taskRun(workerTask.getTaskRun()).build())
                 .toList();
             executorToReturn = executorToReturn.withWorkerTaskResults(failed, "handleWorkerTask");
         }
 
-        // Handle WorkerTasks for CREATED TaskRun
-        List<WorkerTask> workerTasksCreated = workerTasks.get(false); // is not FAILED
-        if (workerTasksCreated != null) {
-            executorToReturn = executorToReturn.withWorkerTasks(workerTasksCreated, "handleWorkerTask");
+        // Send other TaskRun to the worker (create worker tasks)
+        List<WorkerTask> processingTasks = workerTasks.get(false);
+        if (processingTasks != null && !processingTasks.isEmpty()) {
+            executorToReturn = executorToReturn.withWorkerTasks(processingTasks, "handleWorkerTask");
         }
 
         return executorToReturn;
diff --git a/core/src/main/java/io/kestra/core/schedulers/AbstractScheduler.java b/core/src/main/java/io/kestra/core/schedulers/AbstractScheduler.java
index c5106e282..e29d19ade 100644
--- a/core/src/main/java/io/kestra/core/schedulers/AbstractScheduler.java
+++ b/core/src/main/java/io/kestra/core/schedulers/AbstractScheduler.java
@@ -822,7 +822,7 @@ public abstract class AbstractScheduler implements Scheduler, Service {
             .conditionContext(flowWithTriggerWithDefault.conditionContext)
             .build();
         try {
-            this.workerTaskQueue.emit(workerGroupService.resolveGroupFromJob(workerTrigger), workerTrigger);
+            this.workerTaskQueue.emit(workerGroupService.resolveGroupFromJob(workerTrigger).map(group -> group.getKey()).orElse(null), workerTrigger);
         } catch (QueueException e) {
             log.error("Unable to emit the Worker Trigger job", e);
         }
diff --git a/core/src/main/java/io/kestra/core/services/WorkerGroupService.java b/core/src/main/java/io/kestra/core/services/WorkerGroupService.java
index c7795f8eb..b5366c55e 100644
--- a/core/src/main/java/io/kestra/core/services/WorkerGroupService.java
+++ b/core/src/main/java/io/kestra/core/services/WorkerGroupService.java
@@ -1,10 +1,13 @@
 package io.kestra.core.services;
 
+import io.kestra.core.models.tasks.WorkerGroup;
 import io.kestra.core.runners.WorkerJob;
-import io.kestra.core.runners.WorkerTask;
 import jakarta.inject.Singleton;
+import jakarta.validation.Valid;
 import lombok.extern.slf4j.Slf4j;
 
+import java.util.Optional;
+
 /**
  * Provides business logic to manipulate Worker Groups.
  */
@@ -16,9 +19,9 @@ public class WorkerGroupService {
         return null;
     }
 
-    public String resolveGroupFromJob(WorkerJob workerJob) {
+    public Optional<WorkerGroup> resolveGroupFromJob(WorkerJob workerJob) {
         // Worker Group is an EE functionality, setting a worker group in a task is not possible (validation error),
-        // and even if possible it will be a no-op.
-        return null;
+        // and even if possible, it will be a no-op.
+        return Optional.empty();
     }
 }
diff --git a/core/src/main/java/io/kestra/plugin/core/flow/If.java b/core/src/main/java/io/kestra/plugin/core/flow/If.java
index fbf016dbd..25975f43e 100644
--- a/core/src/main/java/io/kestra/plugin/core/flow/If.java
+++ b/core/src/main/java/io/kestra/plugin/core/flow/If.java
@@ -137,11 +137,14 @@ public class If extends Task implements FlowableTask<If.Output> {
         // We need to evaluate the condition once, so if the condition is impacted during the processing or a branch, the same branch is always taken.
         // This can exist for ex if the condition is based on a KV and the KV is changed in the branch.
         // For this, we evaluate the condition in the outputs() method and get it from the outputs.
+        // But unfortunately, the output may not have yet been computed in some cases, like if the task is inside a flowable, in this case we compute the result anyway.
+        Boolean evaluationResult;
         if (parentTaskRun.getOutputs() == null || parentTaskRun.getOutputs().get("evaluationResult") == null) {
-            throw new IllegalVariableEvaluationException("Unable to find the 'evaluationResult' output, this may indicate that the condition evaluation fail, check your execution logs for more information.");
+            evaluationResult = isTrue(runContext);
+        } else {
+            evaluationResult = (Boolean) parentTaskRun.getOutputs().get("evaluationResult");
         }
 
-        Boolean evaluationResult = (Boolean) parentTaskRun.getOutputs().get("evaluationResult");
         if (Boolean.TRUE.equals(evaluationResult)) {
             return FlowableUtils.resolveTasks(then, parentTaskRun);
         }
@@ -178,12 +181,16 @@ public class If extends Task implements FlowableTask<If.Output> {
     }
 
     @Override
-     public If.Output outputs(RunContext runContext) throws Exception {
-        String rendered = runContext.render(condition);
-        boolean evaluationResult = TruthUtils.isTruthy(rendered);
+    public If.Output outputs(RunContext runContext) throws Exception {
+        Boolean evaluationResult = isTrue(runContext);
         return If.Output.builder().evaluationResult(evaluationResult).build();
     }
 
+    private Boolean isTrue(RunContext runContext) throws IllegalVariableEvaluationException {
+        String rendered = runContext.render(condition);
+        return TruthUtils.isTruthy(rendered);
+    }
+
     @Builder
     @Getter
     public static class Output implements io.kestra.core.models.tasks.Output {
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 3f9d215a0..41a974590 100644
--- a/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java
+++ b/jdbc/src/main/java/io/kestra/jdbc/runner/JdbcExecutor.java
@@ -436,7 +436,7 @@ public class JdbcExecutor implements ExecutorInterface, Service {
                                 }
                                 else {
                                     if (workerTask.getTask().isSendToWorkerTask()) {
-                                        workerTaskQueue.emit(workerGroupService.resolveGroupFromJob(workerTask), workerTask);
+                                        workerTaskQueue.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)));
diff --git a/ui/src/assets/documentations/basic.md b/ui/src/assets/documentations/basic.md
index 26b4d0e16..9b35f18e9 100644
--- a/ui/src/assets/documentations/basic.md
+++ b/ui/src/assets/documentations/basic.md
@@ -44,17 +44,17 @@ Note that you need an active Internet connection to view that documentation, as
 
 The following core properties are available in all tasks.
 
-| Property       | Description                                                                                                                                                                           |
-|----------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| `id`           | A unique identifier for the task                                                                                                                                                      |
-| `type`         | A full Java class name that represents the type of task                                                                                                                               |
-| `description`  | Your custom [documentation](https://kestra.io/docs/workflow-components/descriptions) of what the task does                                                                            |
-| `retry`        | How often should the task be retried in case of a failure, and the [type of retry strategy](https://kestra.io/docs/workflow-components/retries)                                       |
-| `timeout`      | The [maximum time allowed](https://kestra.io/docs/workflow-components/timeout) for the task to complete                                                                               |
-| `disabled`     | A boolean flag indicating whether the task is [disabled or not](https://kestra.io/docs/workflow-components/disabled); if set to `true`, the task will be skipped during the execution |
-| `workerGroup`  | The [group of workers](https://kestra.io/docs/enterprise/worker-group) that are eligible to execute the task; you can specify a `workerGroup.key`                                     |
-| `allowFailure` | A boolean flag allowing to continue the execution even if this task fails                                                                                                             |
-| `logLevel`     | The level of log detail to be stored.                                                                                                                                                 |
+| Property       | Description                                                                                                                                                                                                                                                                                                                                                   |
+|----------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| `id`           | A unique identifier for the task                                                                                                                                                                                                                                                                                                                              |
+| `type`         | A full Java class name that represents the type of task                                                                                                                                                                                                                                                                                                       |
+| `description`  | Your custom [documentation](https://kestra.io/docs/workflow-components/descriptions) of what the task does                                                                                                                                                                                                                                                    |
+| `retry`        | How often should the task be retried in case of a failure, and the [type of retry strategy](https://kestra.io/docs/workflow-components/retries)                                                                                                                                                                                                               |
+| `timeout`      | The [maximum time allowed](https://kestra.io/docs/workflow-components/timeout) for the task to complete                                                                                                                                                                                                                                                       |
+| `disabled`     | A boolean flag indicating whether the task is [disabled or not](https://kestra.io/docs/workflow-components/disabled); if set to `true`, the task will be skipped during the execution                                                                                                                                                                         |
+| `workerGroup`  | The [group of workers](https://kestra.io/docs/enterprise/worker-group) that are eligible to execute the task; you can specify a `workerGroup.key` and a `workerGroup.fallback` (by default WAIT)                                                                                                                                                              |
+| `allowFailure` | A boolean flag allowing to continue the execution even if this task fails                                                                                                                                                                                                                                                                                     |
+| `logLevel`     | The level of log detail to be stored.                                                                                                                                                                                                                                                                                                                         |
 | `logToFile`     | A boolean that lets you store logs as a file in internal storage. That file can be previewed and downloaded from the Logs and Gantt Execution tabs. When set to true, logs aren’t saved in the database, which is useful for tasks that produce a large amount of logs that would otherwise take up too much space. The same property can be set on triggers. |
 
 
diff --git a/ui/src/components/executions/Overview.vue b/ui/src/components/executions/Overview.vue
index d6f8a5339..25f53a5d1 100644
--- a/ui/src/components/executions/Overview.vue
+++ b/ui/src/components/executions/Overview.vue
@@ -1,13 +1,13 @@
 <template>
     <div v-if="execution" class="execution-overview">
-        <div v-if="execution.state.current === 'FAILED'" class="error-container">
+        <div v-if="execution.error" class="error-container">
             <div class="error-header" @click="isExpanded = !isExpanded">
                 <svg xmlns="http://www.w3.org/2000/svg" class="error-icon" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2" stroke-linecap="round">
                     <path d="M10.29 3.86L1.82 18a2 2 0 0 0 1.71 3h16.94a2 2 0 0 0 1.71-3L13.71 3.86a2 2 0 0 0-3.42 0z" />
                     <line x1="12" y1="9" x2="12" y2="13" />
                     <line x1="12" y1="17" x2="12.01" y2="17" />
                 </svg>
-                <span class="error-message">{{ errorMessage }}</span>
+                <span class="error-message">{{ execution.error.message }}</span>
                 <span class="toggle-icon">
                     <svg v-if="isExpanded" xmlns="http://www.w3.org/2000/svg" viewBox="0 0 24 24" fill="currentColor" class="arrow-icon">
                         <path d="M18 15l-6-6-6 6" />
@@ -18,7 +18,7 @@
                 </span>
             </div>
             <div v-if="isExpanded" class="error-stack">
-                <div v-for="(line, index) in stackTrace.split('\n')" :key="index" class="stack-line">
+                <div v-for="(line, index) in execution.error.stacktrace?.split('\n')" :key="index" class="stack-line">
                     {{ line }}
                 </div>
             </div>
@@ -179,19 +179,6 @@
                         this.$route.params
                     );
                 }
-            },
-            execution: {
-                handler(newExecution) {
-                    if (newExecution?.error) {
-                        this.errorMessage = newExecution.error.message || "";
-                        this.stackTrace = newExecution.error.stacktrace || [];
-                    }
-                    else {
-                        this.errorMessage = "";
-                        this.stackTrace = [];
-                    }
-                },
-                immediate: true
             }
         },
         data() {
diff --git a/ui/src/translations/de.json b/ui/src/translations/de.json
index a2b4505b9..c4ec403e5 100644
--- a/ui/src/translations/de.json
+++ b/ui/src/translations/de.json
@@ -909,6 +909,8 @@
     "executions paused": "<code>{executionCount}</code> Ausführung(en) PAUSED",
     "executions unqueue": "<code>{executionCount}</code> Ausführung(en) aus der Warteschlange entfernen",
     "unqueue confirm": "Sind Sie sicher, dass Sie die Ausführung <code>{id}</code> aus der Warteschlange entfernen möchten?",
-    "execution-warn-title": "Wichtige Warnung!"
+    "execution-warn-title": "Wichtige Warnung!",
+    "worker group key": "Worker-Gruppe {key}",
+    "worker group fallback": "Arbeitsgruppe Fallback"
   }
 }
\ No newline at end of file
diff --git a/ui/src/translations/en.json b/ui/src/translations/en.json
index 2364fa885..6cdecca8f 100644
--- a/ui/src/translations/en.json
+++ b/ui/src/translations/en.json
@@ -568,6 +568,8 @@
     "flow deleted, you can restore it": "The flow has been deleted and this is a read-only view. You can still restore it.",
     "workers": "Workers",
     "worker group": "Worker Group",
+    "worker group key": "Worker Group key",
+    "worker group fallback": "Worker Group fallback",
     "version": "Version",
     "services": "Services",
     "cluster": "Cluster",
diff --git a/ui/src/translations/es.json b/ui/src/translations/es.json
index 9b922eaab..2cddb3e2a 100644
--- a/ui/src/translations/es.json
+++ b/ui/src/translations/es.json
@@ -909,6 +909,8 @@
     "executions paused": "<code>{executionCount}</code> ejecución(es) PAUSED",
     "executions unqueue": "<code>{executionCount}</code> ejecución(es) descoladas",
     "unqueue confirm": "¿Está seguro de descolar la ejecución <code>{id}</code>?",
-    "execution-warn-title": "¡Advertencia Importante!"
+    "execution-warn-title": "¡Advertencia Importante!",
+    "worker group key": "Grupo de Worker key",
+    "worker group fallback": "Grupo de Worker de respaldo"
   }
 }
\ No newline at end of file
diff --git a/ui/src/translations/fr.json b/ui/src/translations/fr.json
index b782e048a..ea6e154e2 100644
--- a/ui/src/translations/fr.json
+++ b/ui/src/translations/fr.json
@@ -909,6 +909,8 @@
     "executions paused": "<code>{executionCount}</code> exécution(s) PAUSED",
     "executions unqueue": "<code>{executionCount}</code> exécution(s) en file d'attente",
     "unqueue confirm": "Êtes-vous sûr de vouloir retirer l'exécution <code>{id}</code> de la file d'attente ?",
-    "execution-warn-title": "Avertissement Important !"
+    "execution-warn-title": "Avertissement Important !",
+    "worker group key": "Clé du groupe de Worker",
+    "worker group fallback": "Groupe de Worker de secours"
   }
 }
\ No newline at end of file
diff --git a/ui/src/translations/hi.json b/ui/src/translations/hi.json
index 79f76d429..85416709b 100644
--- a/ui/src/translations/hi.json
+++ b/ui/src/translations/hi.json
@@ -909,6 +909,8 @@
     "executions paused": "<code>{executionCount}</code> execution(s) PAUSED",
     "executions unqueue": "<code>{executionCount}</code> execution(s) कतार से बाहर",
     "unqueue confirm": "क्या आप वाकई निष्पादन <code>{id}</code> को अनक्यू करने के लिए सुनिश्चित हैं?",
-    "execution-warn-title": "महत्वपूर्ण WARNING!"
+    "execution-warn-title": "महत्वपूर्ण WARNING!",
+    "worker group key": "वर्कर ग्रुप key",
+    "worker group fallback": "वर्कर ग्रुप फॉलबैक"
   }
 }
\ No newline at end of file
diff --git a/ui/src/translations/it.json b/ui/src/translations/it.json
index 1eae3d827..60a5c86a8 100644
--- a/ui/src/translations/it.json
+++ b/ui/src/translations/it.json
@@ -909,6 +909,8 @@
     "executions paused": "<code>{executionCount}</code> esecuzione(i) PAUSED",
     "executions unqueue": "<code>{executionCount}</code> esecuzione(i) in coda",
     "unqueue confirm": "Sei sicuro di voler rimuovere dalla coda l'esecuzione <code>{id}</code>?",
-    "execution-warn-title": "Avviso Importante!"
+    "execution-warn-title": "Avviso Importante!",
+    "worker group key": "Chiave del gruppo di Worker",
+    "worker group fallback": "Gruppo di Worker di riserva"
   }
 }
\ No newline at end of file
diff --git a/ui/src/translations/ja.json b/ui/src/translations/ja.json
index 9aae6b1a5..82d646520 100644
--- a/ui/src/translations/ja.json
+++ b/ui/src/translations/ja.json
@@ -909,6 +909,8 @@
     "executions paused": "<code>{executionCount}</code> 回の実行がPAUSEDされました",
     "executions unqueue": "<code>{executionCount}</code> 回の実行がキューから外れました",
     "unqueue confirm": "実行 <code>{id}</code> をキューから外してもよろしいですか？",
-    "execution-warn-title": "重要なWarning!"
+    "execution-warn-title": "重要なWarning!",
+    "worker group key": "ワーカーグループキー",
+    "worker group fallback": "ワーカーグループフォールバック"
   }
 }
\ No newline at end of file
diff --git a/ui/src/translations/ko.json b/ui/src/translations/ko.json
index 9bfdd64f1..c4c86c9c7 100644
--- a/ui/src/translations/ko.json
+++ b/ui/src/translations/ko.json
@@ -909,6 +909,8 @@
     "executions paused": "<code>{executionCount}</code> execution(s) PAUSED",
     "executions unqueue": "<code>{executionCount}</code> 실행 대기열에서 제거됨",
     "unqueue confirm": "실행 <code>{id}</code>의 대기열을 해제하시겠습니까?",
-    "execution-warn-title": "중요한 WARNING!"
+    "execution-warn-title": "중요한 WARNING!",
+    "worker group key": "작업자 그룹 key",
+    "worker group fallback": "워크 그룹 폴백"
   }
 }
\ No newline at end of file
diff --git a/ui/src/translations/pl.json b/ui/src/translations/pl.json
index 3642e27b0..14d72b3b8 100644
--- a/ui/src/translations/pl.json
+++ b/ui/src/translations/pl.json
@@ -909,6 +909,8 @@
     "executions paused": "<code>{executionCount}</code> wykonanie(a) wstrzymane",
     "executions unqueue": "<code>{executionCount}</code> wykonanie(a) w kolejce",
     "unqueue confirm": "Czy na pewno chcesz usunąć z kolejki wykonanie <code>{id}</code>?",
-    "execution-warn-title": "Ważne Ostrzeżenie!"
+    "execution-warn-title": "Ważne Ostrzeżenie!",
+    "worker group key": "Klucz grupy Worker",
+    "worker group fallback": "Grupa Workerów zapasowa"
   }
 }
\ No newline at end of file
diff --git a/ui/src/translations/pt.json b/ui/src/translations/pt.json
index ab5cfca32..1a171c199 100644
--- a/ui/src/translations/pt.json
+++ b/ui/src/translations/pt.json
@@ -909,6 +909,8 @@
     "executions paused": "<code>{executionCount}</code> execução(ões) PAUSED",
     "executions unqueue": "<code>{executionCount}</code> execução(ões) desfiladas",
     "unqueue confirm": "Tem certeza de que deseja desagendar a execução <code>{id}</code>?",
-    "execution-warn-title": "Aviso Importante!"
+    "execution-warn-title": "Aviso Importante!",
+    "worker group key": "Chave do Grupo de Worker",
+    "worker group fallback": "Grupo de Worker de reserva"
   }
 }
\ No newline at end of file
diff --git a/ui/src/translations/ru.json b/ui/src/translations/ru.json
index fe934a3a2..df35a7062 100644
--- a/ui/src/translations/ru.json
+++ b/ui/src/translations/ru.json
@@ -909,6 +909,8 @@
     "executions paused": "<code>{executionCount}</code> выполнение(я) PAUSED",
     "executions unqueue": "<code>{executionCount}</code> выполнение(я) в очереди",
     "unqueue confirm": "Вы уверены, что хотите снять выполнение из очереди <code>{id}</code>?",
-    "execution-warn-title": "Важное предупреждение!"
+    "execution-warn-title": "Важное предупреждение!",
+    "worker group key": "Ключ группы Worker",
+    "worker group fallback": "Группа Worker резервный вариант"
   }
 }
\ No newline at end of file
diff --git a/ui/src/translations/zh_CN.json b/ui/src/translations/zh_CN.json
index 91e2c896f..d9c0d77e0 100644
--- a/ui/src/translations/zh_CN.json
+++ b/ui/src/translations/zh_CN.json
@@ -909,6 +909,8 @@
     "executions paused": "<code>{executionCount}</code> 次执行已暂停",
     "executions unqueue": "<code>{executionCount}</code> 次执行已出队",
     "unqueue confirm": "您确定要取消排队执行<code>{id}</code>吗？",
-    "execution-warn-title": "重要警告！"
+    "execution-warn-title": "重要警告！",
+    "worker group key": "工作组 key",
+    "worker group fallback": "工作组回退"
   }
 }
\ No newline at end of file


Test patch:
diff --git a/core/src/test/java/io/kestra/core/docs/ClassPluginDocumentationTest.java b/core/src/test/java/io/kestra/core/docs/ClassPluginDocumentationTest.java
index 34dca000a..b80bff867 100644
--- a/core/src/test/java/io/kestra/core/docs/ClassPluginDocumentationTest.java
+++ b/core/src/test/java/io/kestra/core/docs/ClassPluginDocumentationTest.java
@@ -105,7 +105,7 @@ class ClassPluginDocumentationTest {
             assertThat(doc.getDocLicense(), nullValue());
 
             assertThat(((Map<String, Object>) doc.getDefs().get("io.kestra.core.models.tasks.WorkerGroup")).get("type"), is("object"));
-            assertThat(((Map<String, Object>) ((Map<String, Object>) doc.getDefs().get("io.kestra.core.models.tasks.WorkerGroup")).get("properties")).size(), is(1));
+            assertThat(((Map<String, Object>) ((Map<String, Object>) doc.getDefs().get("io.kestra.core.models.tasks.WorkerGroup")).get("properties")).size(), is(2));
         }));
     }
 
diff --git a/core/src/test/java/io/kestra/core/schedulers/AbstractSchedulerTest.java b/core/src/test/java/io/kestra/core/schedulers/AbstractSchedulerTest.java
index 2c6d8d269..822aab757 100644
--- a/core/src/test/java/io/kestra/core/schedulers/AbstractSchedulerTest.java
+++ b/core/src/test/java/io/kestra/core/schedulers/AbstractSchedulerTest.java
@@ -45,7 +45,7 @@ abstract public class AbstractSchedulerTest {
         UnitTest schedule = UnitTest.builder()
             .id("sleep")
             .type(UnitTest.class.getName())
-            .workerGroup(workerGroup == null ? null : new WorkerGroup(workerGroup))
+            .workerGroup(workerGroup == null ? null : new WorkerGroup(workerGroup, null))
             .build();
 
         return createFlow(Collections.singletonList(schedule), List.of(
diff --git a/core/src/test/java/io/kestra/core/validations/WorkingDirectoryTest.java b/core/src/test/java/io/kestra/core/validations/WorkingDirectoryTest.java
index 6d47cb646..898c7bb79 100644
--- a/core/src/test/java/io/kestra/core/validations/WorkingDirectoryTest.java
+++ b/core/src/test/java/io/kestra/core/validations/WorkingDirectoryTest.java
@@ -75,7 +75,7 @@ public class WorkingDirectoryTest {
                     .id("log")
                     .type(Log.class.getName())
                     .message("Hello World")
-                    .workerGroup(new WorkerGroup("toto"))
+                    .workerGroup(new WorkerGroup("toto", null))
                     .build()
                 )
             )
diff --git a/core/src/test/java/io/kestra/plugin/core/flow/IfTest.java b/core/src/test/java/io/kestra/plugin/core/flow/IfTest.java
index 8de9902ab..9cf3e40ea 100644
--- a/core/src/test/java/io/kestra/plugin/core/flow/IfTest.java
+++ b/core/src/test/java/io/kestra/plugin/core/flow/IfTest.java
@@ -86,4 +86,14 @@ class IfTest  extends AbstractMemoryRunnerTest {
         assertThat(execution.findTaskRunsByTaskId("when-true").isEmpty(), is(true));
         assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
     }
+
+    @Test
+    void ifInFlowable() throws TimeoutException, QueueException {
+        Execution execution = runnerUtils.runOne(null, "io.kestra.tests", "if-in-flowable", null,
+            (f, e) -> Map.of("param", true) , Duration.ofSeconds(120));
+
+        assertThat(execution.getTaskRunList(), hasSize(8));
+        assertThat(execution.findTaskRunsByTaskId("after_if").getFirst().getState().getCurrent(), is(State.Type.SUCCESS));
+        assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
+    }
 }
\ No newline at end of file
diff --git a/core/src/test/resources/flows/valids/if-in-flowable.yaml b/core/src/test/resources/flows/valids/if-in-flowable.yaml
new file mode 100644
index 000000000..8a8f6fe0c
--- /dev/null
+++ b/core/src/test/resources/flows/valids/if-in-flowable.yaml
@@ -0,0 +1,22 @@
+id: if-in-flowable
+namespace: io.kestra.tests
+
+inputs:
+  - id: param
+    type: STRING
+
+tasks:
+  - id: for_each
+    type: io.kestra.plugin.core.flow.ForEach
+    values: ["value 1", "value 2", "value 3"]
+    tasks:
+      - id: before_if
+        type: io.kestra.plugin.core.debug.Return
+        format: "Before if: {{ taskrun.value }}"
+      - id: if
+        type: io.kestra.plugin.core.flow.If
+        condition: "{{ taskrun.value equals 'value 2' }}"
+        then:
+          - id: after_if
+            type: io.kestra.plugin.core.debug.Return
+            format: "After if: {{ parent.taskrun.value }}"
\ No newline at end of file