Repository: kestra-io/kestra (java)

Issue:
ForEachItem fails with [Cannot invoke "java.util.Map.get(Object)" because "taskOutput" is null] when there are no items to process
TLDR:

Here is a simple workflow you can use to reproduce the problem:

```
id: debug_each_issue
namespace: kestra_issues

tasks:

  - id: workingDir
    type: io.kestra.plugin.core.flow.WorkingDirectory
    tasks:
    - id: bash
      type: io.kestra.plugin.scripts.shell.Commands
      commands:
        - touch test
    - id: outputFiles
      type: io.kestra.plugin.core.storage.LocalFiles
      outputs:
        - test

  - id: for_each_will_fail
    type: io.kestra.plugin.core.flow.ForEachItem
    items: "{{ outputs.outputFiles.uris['test'] }}"
    batch:
      partitions: 5 
    flowId: debug_each_subflow
    namespace: kestra_issues
    wait: true
    transmitFailed: true
    inputs:
      test: "{{ taskrun.items }}"
```

```
id: debug_each_subflow
namespace: kestra_issues

inputs:
  - id: test
    type: STRING

tasks:

  - id: print_status
    type: io.kestra.plugin.core.log.Log
    message: got {{ inputs.test}}
```

<img width="1358" alt="image" src="https://github.com/kestra-io/kestra/assets/678596/ffc9acc4-4cde-4b80-8e4d-a212a47bc972">

---

### Describe the issue

UPD: it looks like the issue happens when there is no items.
I used FilteTransform that filters out some records. Probably ForEachItem did not have data to process
Still this is not expected behaviour. If there are no items to process FEI should complete without errors

---

When Kafka trigger returns only 1 record the following flow fails with 

```java.util.Map.get(Object)" because "taskOutput" is null```

Also, notice that it won't create a subflow execution.

When Kafka trigger returns multiple records, everything works.

flow:

```
....

concurrency:
  limit: 1 

tasks:
  - id: select_suppliers
    type: io.kestra.plugin.scripts.jython.FileTransform    
    from: "{{ trigger.uri }}"
    concurrent: 3
    script: |
         [rest is omitted for brevity]

  - id: verify_suppliers_in_batches
    type: io.kestra.plugin.core.flow.ForEachItem
    items: "{{ outputs.select_suppliers.uri }}"
    batch:
      partitions: 5 
    flowId: [SUBFLOW ID]
    namespace: [REDACTED]
    wait: true
    transmitFailed: true
    inputs:
      suppliers: "{{ taskrun.items }}"

  - id: purge_execution
    type: io.kestra.plugin.core.storage.PurgeExecution

pluginDefaults:
  - type: io.kestra.plugin.kafka.Trigger
    values:
      logLevel: WARN

triggers:
  - id: read_suppliers_registrations
    [rest is omitted for brevity]    
```

Where there is 1 record:

<img width="867" alt="image" src="https://github.com/kestra-io/kestra/assets/678596/afb409db-b210-4e16-a789-2e0e3bebd89d">

<img width="721" alt="image" src="https://github.com/kestra-io/kestra/assets/678596/264527c5-abe6-4d65-8c72-3c04dca31e72">

When there are > 1 records

<img width="876" alt="image" src="https://github.com/kestra-io/kestra/assets/678596/4d19104b-4638-4b88-b27d-70115b1c4611">

<img width="1347" alt="image" src="https://github.com/kestra-io/kestra/assets/678596/64827405-e0df-41a8-903e-e075b52fb3cd">


### Environment

- Kestra Version: 17.8
- Operating System (OS/Docker/Kubernetes):
- Java Version (if you don't run kestra in Docker):


Pull request description:
fix(core): ForEachItem NPE when no items
Fixes #4278

Patch:
diff --git a/core/src/main/java/io/kestra/plugin/core/flow/ForEachItem.java b/core/src/main/java/io/kestra/plugin/core/flow/ForEachItem.java
index 6f539c6ba..0272c7283 100644
--- a/core/src/main/java/io/kestra/plugin/core/flow/ForEachItem.java
+++ b/core/src/main/java/io/kestra/plugin/core/flow/ForEachItem.java
@@ -561,6 +561,11 @@ public class ForEachItem extends Task implements FlowableTask<VoidOutput>, Child
             // get the list of splits from the outputs of the split task
             String taskId = this.id.substring(0, this.id.lastIndexOf('_')) + ForEachItemExecutable.SUFFIX;
             var taskOutput = extractOutput(runContext, taskId);
+            if (taskOutput == null) {
+                // there were no subflow executions
+                return null;
+            }
+
             Integer iterations = (Integer) taskOutput.get(ExecutableUtils.TASK_VARIABLE_NUMBER_OF_BATCHES);
             String subflowOutputsBaseUri = (String) taskOutput.get(ExecutableUtils.TASK_VARIABLE_SUBFLOW_OUTPUTS_BASE_URI);
 


Test patch:
diff --git a/core/src/test/java/io/kestra/plugin/core/flow/ForEachItemCaseTest.java b/core/src/test/java/io/kestra/plugin/core/flow/ForEachItemCaseTest.java
index 4fd1d0ff9..5e5187ac5 100644
--- a/core/src/test/java/io/kestra/plugin/core/flow/ForEachItemCaseTest.java
+++ b/core/src/test/java/io/kestra/plugin/core/flow/ForEachItemCaseTest.java
@@ -102,6 +102,21 @@ public class ForEachItemCaseTest {
         assertThat(triggered.get().getTaskRunList(), hasSize(1));
     }
 
+    @SuppressWarnings("unchecked")
+    public void forEachItemEmptyItems() throws TimeoutException, URISyntaxException, IOException {
+        URI file = emptyItems();
+        Map<String, Object> inputs = Map.of("file", file.toString());
+        Execution execution = runnerUtils.runOne(null, TEST_NAMESPACE, "for-each-item", null,
+            (flow, execution1) -> flowIO.typedInputs(flow, execution1, inputs),
+            Duration.ofSeconds(30));
+
+        // assert on the main flow execution
+        assertThat(execution.getTaskRunList(), hasSize(4));
+        assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
+        Map<String, Object> outputs = execution.getTaskRunList().get(2).getOutputs();
+        assertThat(outputs, nullValue());
+    }
+
     @SuppressWarnings("unchecked")
     public void forEachItemNoWait() throws TimeoutException, InterruptedException, URISyntaxException, IOException {
         CountDownLatch countDownLatch = new CountDownLatch(26);
@@ -261,6 +276,16 @@ public class ForEachItemCaseTest {
         );
     }
 
+    private URI emptyItems() throws URISyntaxException, IOException {
+        File tempFile = File.createTempFile("file", ".txt");
+
+        return storageInterface.put(
+            null,
+            new URI("/file/storage/file.txt"),
+            new FileInputStream(tempFile)
+        );
+    }
+
     private List<String> content() {
         return IntStream
             .range(0, 102)
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 4a2cbc92c..1dcd37145 100644
--- a/jdbc/src/test/java/io/kestra/jdbc/runner/JdbcRunnerTest.java
+++ b/jdbc/src/test/java/io/kestra/jdbc/runner/JdbcRunnerTest.java
@@ -276,6 +276,11 @@ public abstract class JdbcRunnerTest {
         forEachItemCaseTest.forEachItem();
     }
 
+    @Test
+    void forEachItemEmptyItems() throws URISyntaxException, IOException, TimeoutException {
+        forEachItemCaseTest.forEachItemEmptyItems();
+    }
+
     @Test
     void forEachItemNoWait() throws URISyntaxException, IOException, InterruptedException, TimeoutException {
         forEachItemCaseTest.forEachItemNoWait();