Repository: kestra-io/kestra
Issue: Handle null value on Each
Labels: bug
backend

if you are using evaluation on Each `value`, sometime there can have a `null`.

You don't want this ! 
There will be `value` set to `[null]` and on child task `{{ taskrun.value }}` will not exist (the serialization will remove it).

If you must raise an exception if only null values

Issue comments:

Pull request review comments:

Patch that resolved the issue:
diff --git a/core/src/main/java/org/kestra/core/exceptions/IllegalVariableEvaluationException.java b/core/src/main/java/org/kestra/core/exceptions/IllegalVariableEvaluationException.java
index d5ce098205..9844269478 100644
--- a/core/src/main/java/org/kestra/core/exceptions/IllegalVariableEvaluationException.java
+++ b/core/src/main/java/org/kestra/core/exceptions/IllegalVariableEvaluationException.java
@@ -1,6 +1,6 @@
 package org.kestra.core.exceptions;
 
-public class IllegalVariableEvaluationException extends Exception {
+public class IllegalVariableEvaluationException extends InternalException {
     public IllegalVariableEvaluationException(Throwable e) {
         super(e);
     }
diff --git a/core/src/main/java/org/kestra/core/runners/AbstractExecutor.java b/core/src/main/java/org/kestra/core/runners/AbstractExecutor.java
index 94e502bd23..af66c71b90 100644
--- a/core/src/main/java/org/kestra/core/runners/AbstractExecutor.java
+++ b/core/src/main/java/org/kestra/core/runners/AbstractExecutor.java
@@ -82,7 +82,7 @@ protected Execution onNexts(Flow flow, Execution execution, List<TaskRun> nexts)
         return newExecution;
     }
 
-    private Optional<WorkerTaskResult> childWorkerTaskResult(Flow flow, Execution execution, TaskRun parentTaskRun) throws IllegalVariableEvaluationException, InternalException {
+    private Optional<WorkerTaskResult> childWorkerTaskResult(Flow flow, Execution execution, TaskRun parentTaskRun) throws InternalException {
         Task parent = flow.findTaskByTaskId(parentTaskRun.getTaskId());
 
         if (parent instanceof FlowableTask) {
@@ -157,7 +157,7 @@ private Optional<WorkerTaskResult> childWorkerTaskTypeToWorkerTask(
             .findFirst();
     }
 
-    private Optional<List<TaskRun>> childNextsTaskRun(Flow flow, Execution execution, TaskRun parentTaskRun) throws IllegalVariableEvaluationException, InternalException {
+    private Optional<List<TaskRun>> childNextsTaskRun(Flow flow, Execution execution, TaskRun parentTaskRun) throws InternalException {
         Task parent = flow.findTaskByTaskId(parentTaskRun.getTaskId());
 
         if (parent instanceof FlowableTask) {
@@ -265,7 +265,7 @@ protected Optional<Execution> doMain(Execution execution, Flow flow) {
         return this.handleKilling(execution, flow);
     }
 
-    protected Optional<List<TaskRun>> doNexts(Execution execution, Flow flow) throws Exception {
+    protected Optional<List<TaskRun>> doNexts(Execution execution, Flow flow) throws InternalException {
         List<TaskRun> nexts;
 
         // killing, so no more nexts
@@ -290,7 +290,7 @@ protected Optional<List<TaskRun>> doNexts(Execution execution, Flow flow) throws
         return Optional.empty();
     }
 
-    protected Optional<List<WorkerTask>> doWorkerTask(Execution execution, Flow flow) throws Exception {
+    protected Optional<List<WorkerTask>> doWorkerTask(Execution execution, Flow flow) throws InternalException {
         List<WorkerTask> nexts;
 
         nexts = this.handleWorkerTask(execution, flow);
@@ -301,7 +301,7 @@ protected Optional<List<WorkerTask>> doWorkerTask(Execution execution, Flow flow
         return Optional.empty();
     }
 
-    protected Optional<List<WorkerTaskResult>> doWorkerTaskResult(Execution execution, Flow flow) throws Exception {
+    protected Optional<List<WorkerTaskResult>> doWorkerTaskResult(Execution execution, Flow flow) throws InternalException {
         List<WorkerTaskResult> nexts;
 
         nexts = this.handleChildWorkerTaskResult(execution, flow);
@@ -326,23 +326,31 @@ private List<TaskRun> handleNext(Execution execution, Flow flow) {
         );
     }
 
-    private List<TaskRun> handleChildNext(Execution execution, Flow flow) throws Exception {
+    private List<TaskRun> handleChildNext(Execution execution, Flow flow) throws InternalException {
         if (execution.getTaskRunList() == null) {
             return new ArrayList<>();
         }
 
-        return execution
+
+        List<TaskRun> running = execution
             .getTaskRunList()
             .stream()
             .filter(taskRun -> taskRun.getState().isRunning())
-            .flatMap(throwFunction(taskRun -> this.childNextsTaskRun(flow, execution, taskRun)
-                .orElse(new ArrayList<>())
-                .stream()
-            ))
             .collect(Collectors.toList());
+
+        // Remove functionnal style to avoid (class org.kestra.core.exceptions.IllegalVariableEvaluationException cannot be cast to class java.lang.RuntimeException'
+        ArrayList<TaskRun> result = new ArrayList<>();
+
+        for (TaskRun taskRun :running) {
+            result.addAll(this.childNextsTaskRun(flow, execution, taskRun)
+                .orElse(new ArrayList<>())
+            );
+        }
+
+        return result;
     }
 
-    private List<WorkerTaskResult> handleChildWorkerTaskResult(Execution execution, Flow flow) throws Exception {
+    private List<WorkerTaskResult> handleChildWorkerTaskResult(Execution execution, Flow flow) throws InternalException {
         if (execution.getTaskRunList() == null) {
             return new ArrayList<>();
         }
@@ -412,7 +420,7 @@ private Optional<Execution> handleKilling(Execution execution, Flow flow) {
     }
 
 
-    private List<WorkerTask> handleWorkerTask(Execution execution, Flow flow) throws Exception {
+    private List<WorkerTask> handleWorkerTask(Execution execution, Flow flow) throws InternalException {
         if (execution.getTaskRunList() == null || execution.getState().getCurrent() == State.Type.KILLING) {
             return new ArrayList<>();
         }
diff --git a/core/src/main/java/org/kestra/core/runners/FlowableUtils.java b/core/src/main/java/org/kestra/core/runners/FlowableUtils.java
index 9bfbeef26a..be07caa3ca 100644
--- a/core/src/main/java/org/kestra/core/runners/FlowableUtils.java
+++ b/core/src/main/java/org/kestra/core/runners/FlowableUtils.java
@@ -185,6 +185,17 @@ public static List<ResolvedTask> resolveEachTasks(RunContext runContext, TaskRun
             throw new IllegalVariableEvaluationException(e);
         }
 
+        long nullCount = Arrays
+            .stream(values)
+            .filter(Objects::isNull)
+            .count();
+
+        if (nullCount > 0) {
+            throw new IllegalVariableEvaluationException("Found '" + nullCount + "' null values on Each, " +
+                "with values=" + Arrays.toString(values)
+            );
+        }
+
         return Arrays
             .stream(values)
             .distinct()
diff --git a/runner-memory/src/main/java/org/kestra/runner/memory/MemoryExecutor.java b/runner-memory/src/main/java/org/kestra/runner/memory/MemoryExecutor.java
index 7789bab953..8ac67f9b1f 100644
--- a/runner-memory/src/main/java/org/kestra/runner/memory/MemoryExecutor.java
+++ b/runner-memory/src/main/java/org/kestra/runner/memory/MemoryExecutor.java
@@ -139,7 +139,7 @@ private void handleExecution(ExecutionState state) {
     private void handleFailedExecutionFromExecutor(Execution execution, Exception e) {
         Execution.FailedExecutionWithLog failedExecutionWithLog = execution.failedExecutionFromExecutor(e);
         try {
-            log.error("Failed from executor with {}", e.getMessage(), e);
+            log.error("Failed from executor with '{}'", e.getMessage(), e);
 
             failedExecutionWithLog.getLogs().forEach(logQueue::emit);
 


Test patch:
diff --git a/core/src/test/java/org/kestra/core/Helpers.java b/core/src/test/java/org/kestra/core/Helpers.java
index ada1667b9c..1be6e6e1a1 100644
--- a/core/src/test/java/org/kestra/core/Helpers.java
+++ b/core/src/test/java/org/kestra/core/Helpers.java
@@ -17,7 +17,7 @@
 import java.util.function.BiConsumer;
 
 public class Helpers {
-    public static long FLOWS_COUNT = 38;
+    public static long FLOWS_COUNT = 39;
 
     public static ApplicationContext applicationContext() throws URISyntaxException {
         return applicationContext(Paths.get(Objects.requireNonNull(Helpers.class.getClassLoader().getResource("plugins")).toURI()));
diff --git a/core/src/test/java/org/kestra/core/tasks/flows/EachSequentialTest.java b/core/src/test/java/org/kestra/core/tasks/flows/EachSequentialTest.java
index 2b86fc4a41..efec7479bb 100644
--- a/core/src/test/java/org/kestra/core/tasks/flows/EachSequentialTest.java
+++ b/core/src/test/java/org/kestra/core/tasks/flows/EachSequentialTest.java
@@ -1,30 +1,39 @@
 package org.kestra.core.tasks.flows;
 
+import org.hamcrest.Matchers;
 import org.junit.jupiter.api.Test;
 import org.kestra.core.exceptions.InternalException;
 import org.kestra.core.models.executions.Execution;
+import org.kestra.core.models.executions.LogEntry;
 import org.kestra.core.models.executions.TaskRun;
+import org.kestra.core.models.flows.State;
+import org.kestra.core.queues.QueueFactoryInterface;
+import org.kestra.core.queues.QueueInterface;
 import org.kestra.core.runners.AbstractMemoryRunnerTest;
+import org.kestra.core.runners.RunnerUtils;
 
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.Map;
+import java.util.*;
 import java.util.concurrent.TimeoutException;
 
+import javax.inject.Inject;
+import javax.inject.Named;
+
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.core.StringContains.containsString;
 
 public class EachSequentialTest extends AbstractMemoryRunnerTest {
+    @Inject
+    @Named(QueueFactoryInterface.WORKERTASKLOG_NAMED)
+    QueueInterface<LogEntry> logQueue;
+
     @Test
     void sequential() throws TimeoutException {
         Execution execution = runnerUtils.runOne("org.kestra.tests", "each-sequential");
 
         assertThat(execution.getTaskRunList(), hasSize(8));
-
-        Map<String, Object> outputs = execution.outputs();
-        assertThat(execution.getTaskRunList(), hasSize(8));
+        assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
     }
 
     @Test
@@ -32,6 +41,7 @@ void sequentialNested() throws TimeoutException, InternalException {
         Execution execution = runnerUtils.runOne("org.kestra.tests", "each-sequential-nested");
 
         assertThat(execution.getTaskRunList(), hasSize(23));
+        assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
 
         TaskRun last = execution.findTaskRunsByTaskId("2_return").get(0);
         TaskRun lastWithValue = execution.findTaskRunByTaskIdAndValue("1-2-1_return", Arrays.asList("s1", "a a"));
@@ -52,6 +62,23 @@ void eachEmpty() throws TimeoutException {
         Execution execution = runnerUtils.runOne("org.kestra.tests", "each-empty");
 
         assertThat(execution.getTaskRunList(), hasSize(2));
+        assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
+    }
+
+    @Test
+    void eachNull() throws TimeoutException {
+        EachSequentialTest.eachNullTest(runnerUtils, logQueue);
+    }
+
+    public static void eachNullTest(RunnerUtils runnerUtils, QueueInterface<LogEntry> logQueue) throws TimeoutException {
+        List<LogEntry> logs = new ArrayList<>();
+        logQueue.receive(logs::add);
+
+        Execution execution = runnerUtils.runOne("org.kestra.tests", "each-null");
+
+        assertThat(execution.getTaskRunList(), hasSize(1));
+        assertThat(execution.getState().getCurrent(), is(State.Type.FAILED));
+        assertThat(logs.get(1).getMessage(), Matchers.containsString("Found '1' null values on Each, with values=[1, null, 2]"));
     }
 
     @Test
@@ -59,6 +86,7 @@ void eachSwitch() throws TimeoutException, InternalException {
         Execution execution = runnerUtils.runOne("org.kestra.tests", "each-switch");
 
         assertThat(execution.getTaskRunList(), hasSize(12));
+        assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
 
         TaskRun switchNumber1 = execution.findTaskRunByTaskIdAndValue("2-1-1_switch-number-1", Arrays.asList("b", "1"));
         assertThat((String) switchNumber1.getOutputs().get("value"), is("1"));
diff --git a/core/src/test/resources/flows/valids/each-null.yaml b/core/src/test/resources/flows/valids/each-null.yaml
new file mode 100644
index 0000000000..90a6dde537
--- /dev/null
+++ b/core/src/test/resources/flows/valids/each-null.yaml
@@ -0,0 +1,14 @@
+id: each-null
+namespace: org.kestra.tests
+
+tasks:
+  - id: 1_each
+    type: org.kestra.core.tasks.flows.EachSequential
+    value: '[1, null, 2]'
+    tasks:
+      - id: 1-1
+        type: org.kestra.core.tasks.debugs.Return
+        format: "{{task.id}} > {{taskrun.value}} > {{taskrun.startDate}}"
+  - id: 2_end
+    type: org.kestra.core.tasks.debugs.Return
+    format: "{{task.id}} > {{taskrun.startDate}}"
diff --git a/runner-kafka/src/test/java/org/kestra/runner/kafka/KafkaRunnerTest.java b/runner-kafka/src/test/java/org/kestra/runner/kafka/KafkaRunnerTest.java
index 7bab70f3c8..0476c70241 100644
--- a/runner-kafka/src/test/java/org/kestra/runner/kafka/KafkaRunnerTest.java
+++ b/runner-kafka/src/test/java/org/kestra/runner/kafka/KafkaRunnerTest.java
@@ -14,6 +14,7 @@
 import org.kestra.core.runners.InputsTest;
 import org.kestra.core.runners.ListenersTest;
 import org.kestra.core.runners.RestartCaseTest;
+import org.kestra.core.tasks.flows.EachSequentialTest;
 import org.kestra.core.tasks.flows.TemplateTest;
 import org.kestra.core.utils.TestsUtils;
 
@@ -196,6 +197,11 @@ void flowTrigger() throws Exception {
         flowTriggerCaseTest.trigger();
     }
 
+    @Test
+    void eachWithNull() throws Exception {
+        EachSequentialTest.eachNullTest(runnerUtils, workerTaskLogQueue);
+    }
+
     @Test
     void withTemplate() throws Exception {
         TemplateTest.withTemplate(runnerUtils, templateRepository);
diff --git a/runner-kafka/src/test/resources/application.yml b/runner-kafka/src/test/resources/application.yml
index 1e2def32a9..bafcceee64 100644
--- a/runner-kafka/src/test/resources/application.yml
+++ b/runner-kafka/src/test/resources/application.yml
@@ -73,7 +73,7 @@ kestra:
           cleanup.policy: "delete,compact"
 
       killed:
-        name: "kestra_executor_killed"
+        name: "unit_${kestra.prefix}_kestra_executor_killed"
         cls: org.kestra.core.models.executions.ExecutionKilled
         properties:
           cleanup.policy: "delete,compact"