Repository: kestra-io/kestra (java)

Issue:
When a file provided as `items` property cannot be split into batches, the execution ends in an error without any log message explaining why this task failed
We should raise a friendly error message if the value of the `items` property is not a valid Kestra's internal storage URI. Trying to pass a string that is not a URI doesn't raise any error/warning. The task fails with no information explaining why it failed. It would be great to add extra info for troubleshooting e.g. "It looks like you didn't pass the correct value to the `items` property. Make sure that you set this to an internal storage URI, such as a FILE input or output URI from the previous task."

## Reproducer

subflow:
```yaml
id: orders
namespace: prod

inputs:
  - name: order
    type: STRING

tasks:
  - id: read_file
    type: io.kestra.plugin.scripts.shell.Commands
    runner: PROCESS
    commands:
      - cat "{{ inputs.order }}"

  - id: read_file_content
    type: io.kestra.core.tasks.log.Log
    message: "{{ read(inputs.order) }}"
```

parent: 

```yaml
id: orders_batch
namespace: prod

tasks:
  - id: extract
    type: io.kestra.plugin.fs.http.Download
    uri: https://huggingface.co/datasets/kestra/datasets/resolve/main/json/user_events.json

  - id: each
    type: io.kestra.core.tasks.flows.ForEachItem
    items: "{{ read(outputs.extract.uri) }}" # at runtime, we should give a friendly error message that this is not a valid internal storage file's URI
    batch:
      rows: 1
    namespace: prod
    flowId: orders
    wait: true
    transmitFailed: true
    inputs:
      order: "{{ taskrun.items }}"
```

![image](https://github.com/kestra-io/kestra/assets/86264395/4a12795f-6383-40ed-a9fa-2d042485f4b2)

![image](https://github.com/kestra-io/kestra/assets/86264395/c473f97c-453f-4d3c-91d9-39b1b259ef85)




Pull request description:
feat(core): add error logs when ForEachItem fail to create executions
Fixes #2517

The following task:

```
  - id: each
    type: io.kestra.core.tasks.flows.ForEachItem
    items: "somefile.txt"
    batch: 
      rows: 10
    namespace: dev
    flowId: per-item
    wait: true
    transmitFailed: true
    inputs:
      items: "{{taskrun.items}}"
```

Will generate the following log visible in the UI:

![image](https://github.com/kestra-io/kestra/assets/1819009/f5594e77-4228-4364-8cc6-e7116dbb72f2)



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/core/src/main/java/io/kestra/core/tasks/flows/ForEachItem.java b/core/src/main/java/io/kestra/core/tasks/flows/ForEachItem.java
index a570a4ac8..facd4f718 100644
--- a/core/src/main/java/io/kestra/core/tasks/flows/ForEachItem.java
+++ b/core/src/main/java/io/kestra/core/tasks/flows/ForEachItem.java
@@ -178,8 +178,15 @@ public class ForEachItem extends Task implements ExecutableTask<ForEachItem.Outp
         Execution currentExecution,
         TaskRun currentTaskRun
     ) throws InternalException {
+        var renderedUri = runContext.render(this.items);
+        if (!renderedUri.startsWith("kestra://")) {
+            var errorMessage = "Unable to split the items from " + renderedUri + ", this is not an internal storage URI!";
+            runContext.logger().error(errorMessage);
+            throw new IllegalArgumentException(errorMessage);
+        }
+
         try {
-            List<URI> splits = StorageService.split(runContext, this.batch, URI.create(runContext.render(this.items)));
+            List<URI> splits = StorageService.split(runContext, this.batch, URI.create(renderedUri));
 
             AtomicInteger currentIteration = new AtomicInteger(1);
 
@@ -221,6 +228,7 @@ public class ForEachItem extends Task implements ExecutableTask<ForEachItem.Outp
                 ))
                 .toList();
         } catch (IOException e) {
+            runContext.logger().error(e.getMessage(), e);
             throw new InternalException(e);
         }
     }
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();
+    }
 }