Repository: kestra-io/kestra
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
Labels: bug

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)




Issue comments:
There is no easy fix for this, logs from executable tasks are not sent so even if I add one it would not be sent anywhere so not be visible (I think the same applies to logs from flowable tasks).

As a side note, when an exception occurs that is not caught in a task, it goes into the executor exception handling, which will fail the task and the execution. It will add an error logs to the flow execution that is not visible in the Logs tab of an execution as this one only contains tasks logs. You will be able to see it when looking at the Logs page.
I created https://github.com/kestra-io/kestra/issues/2521 as a followup to the more broad issue of logs related to a flow execution not linked to any tasks.

Pull request review comments:
```suggestion
        if (!renderedUri.startsWith("kestra://")) {
```

Patch that resolved the issue:
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 a570a4ac80..facd4f718e 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 List<WorkerTaskExecution<?>> createWorkerTaskExecutions(
         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 List<WorkerTaskExecution<?>> createWorkerTaskExecutions(
                 ))
                 .toList();
         } catch (IOException e) {
+            runContext.logger().error(e.getMessage(), e);
             throw new InternalException(e);
         }
     }


Test patch: