Repository: kestra-io/kestra (java)

Issue:
Weird if branch behavior
### Describe the issue

Not sure if this is intended, but I didn't find any mention of this behavior in the docs, and it confused me for a bit.

I set up a pretty simple flow that just toggles a KV between "foo" and "bar"

<details><summary>Click to expand</summary>

```yaml
id: myflow
namespace: company.team
tasks:
  - id: if
    type: io.kestra.plugin.core.flow.If
    condition: "{{ kv('myvalue') == 'foo' }}"
    then:
    - id: log-start-foo
      type: io.kestra.plugin.core.log.Log
      message: Starting foo branch
    - id: set-bar
      type: io.kestra.plugin.core.kv.Set
      key: myvalue
      value: bar
    - id: log-end-foo
      type: io.kestra.plugin.core.log.Log
      message: Ending foo branch
    else:
    - id: log-start-bar
      type: io.kestra.plugin.core.log.Log
      message: Starting bar branch
    - id: set-foo
      type: io.kestra.plugin.core.kv.Set
      key: myvalue
      value: foo
    - id: log-end-bar
      type: io.kestra.plugin.core.log.Log
      message: Ending bar branch
 ```

</details>

Here's the topology for it:
![](https://p.cp3.es/ss1728811377-ad4f1c.png) 

From this and the other docs I've read, I would expect this to run in series like a computer program, either running
- if -> log-start-foo -> set-bar -> log-end-foo
or
- if -> log-start-bar -> set-foo -> log-end-bar

However, when I execute this flow it does this:

- if -> log-start-bar -> set-foo -> log-start-foo -> set-bar -> log-end-bar

![](https://p.cp3.es/ss1728811947-7de727.png)

![](https://p.cp3.es/ss1728812405-e03ff9.png)

It seems the "If" task will re-evaluate the condition and switch back and forth between paths as it changes, rather than picking a path and sticking to it.

### Environment

- Kestra Version: 0.19.2 - 4338437 Tue, Oct 8, 2024 9:52 AM
- Operating System (OS/Docker/Kubernetes): Docker (kestra/kestra:latest)
- Java Version (if you don't run kestra in Docker):


Pull request:
fix(core): save the If evaluation condition in the output
This avoids re-evaluation on each child task run which can be an issue if the taskrun modify something that is part of the condition.

Fixes #5437

Patch:
diff --git a/core/src/main/java/io/kestra/core/runners/pebble/functions/CurrentEachOutputFunction.java b/core/src/main/java/io/kestra/core/runners/pebble/functions/CurrentEachOutputFunction.java
index 41d729416..52c8f1ccd 100644
--- a/core/src/main/java/io/kestra/core/runners/pebble/functions/CurrentEachOutputFunction.java
+++ b/core/src/main/java/io/kestra/core/runners/pebble/functions/CurrentEachOutputFunction.java
@@ -27,7 +27,10 @@ public class CurrentEachOutputFunction implements Function {
         if (parents != null && !parents.isEmpty()) {
             Collections.reverse(parents);
             for (Map<?, ?> parent : parents) {
-                outputs = (Map<?, ?>) outputs.get(((Map<?, ?>) parent.get("taskrun")).get("value"));
+                Map<?, ?> taskrun = (Map<?, ?>) parent.get("taskrun");
+                if (taskrun != null) {
+                    outputs = (Map<?, ?>) outputs.get(taskrun.get("value"));
+                }
             }
         }
         Map<?, ?> taskrun = (Map<?, ?>) context.getVariable("taskrun");


Test patch:
diff --git a/core/src/test/java/io/kestra/core/tasks/FetchTest.java b/core/src/test/java/io/kestra/core/tasks/FetchTest.java
index f456aee03..4bb498ba0 100644
--- a/core/src/test/java/io/kestra/core/tasks/FetchTest.java
+++ b/core/src/test/java/io/kestra/core/tasks/FetchTest.java
@@ -3,9 +3,7 @@ package io.kestra.core.tasks;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.executions.TaskRun;
 import io.kestra.core.models.flows.State;
-import io.kestra.core.repositories.FlowRepositoryInterface;
 import io.kestra.core.runners.AbstractMemoryRunnerTest;
-import jakarta.inject.Inject;
 import org.junit.jupiter.api.Test;
 
 import static org.hamcrest.MatcherAssert.assertThat;
@@ -13,17 +11,14 @@ import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.is;
 
 public class FetchTest extends AbstractMemoryRunnerTest {
-    @Inject
-    FlowRepositoryInterface flowRepository;
-
     @Test
     void fetch() throws Exception {
         Execution execution = runnerUtils.runOne(null, "io.kestra.tests", "get-log");
 
         assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
-        assertThat(execution.getTaskRunList(), hasSize(3));
-        TaskRun fetch = execution.getTaskRunList().get(2);
-        assertThat(fetch.getOutputs().get("size"), is(2));
+        assertThat(execution.getTaskRunList(), hasSize(4));
+        TaskRun fetch = execution.findTaskRunsByTaskId("get-log-task").getFirst();
+        assertThat(fetch.getOutputs().get("size"), is(3));
     }
 
     @Test
@@ -31,8 +26,8 @@ public class FetchTest extends AbstractMemoryRunnerTest {
         Execution execution = runnerUtils.runOne(null, "io.kestra.tests", "get-log-taskid");
 
         assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
-        assertThat(execution.getTaskRunList(), hasSize(3));
-        TaskRun fetch = execution.getTaskRunList().get(2);
+        assertThat(execution.getTaskRunList(), hasSize(4));
+        TaskRun fetch = execution.findTaskRunsByTaskId("get-log-task").getFirst();
         assertThat(fetch.getOutputs().get("size"), is(1));
     }
 
@@ -41,8 +36,8 @@ public class FetchTest extends AbstractMemoryRunnerTest {
         Execution execution = runnerUtils.runOne(null, "io.kestra.tests", "get-log-executionid");
 
         assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
-        assertThat(execution.getTaskRunList(), hasSize(3));
-        TaskRun fetch = execution.getTaskRunList().get(2);
-        assertThat(fetch.getOutputs().get("size"), is(2));
+        assertThat(execution.getTaskRunList(), hasSize(4));
+        TaskRun fetch = execution.findTaskRunsByTaskId("get-log-task").getFirst();
+        assertThat(fetch.getOutputs().get("size"), is(3));
     }
 }
diff --git a/core/src/test/resources/flows/valids/get-log-executionid.yaml b/core/src/test/resources/flows/valids/get-log-executionid.yaml
index 3621fc94d..744e02671 100644
--- a/core/src/test/resources/flows/valids/get-log-executionid.yaml
+++ b/core/src/test/resources/flows/valids/get-log-executionid.yaml
@@ -7,6 +7,10 @@ tasks:
   - type: io.kestra.plugin.core.debug.Echo
     id: task-2
     format: task 2
+  # as logs are async, to get your own logs reliably on tasks that run quickly, you must sleep a little
+  - type: io.kestra.plugin.core.flow.Sleep
+    id: sleep
+    duration: PT0.1S
   - type: io.kestra.plugin.core.log.Fetch
     id: get-log-task
     executionId: "{{execution.id}}"
\ No newline at end of file
diff --git a/core/src/test/resources/flows/valids/get-log-taskid.yaml b/core/src/test/resources/flows/valids/get-log-taskid.yaml
index 0b03a2d6d..9f4ca2991 100644
--- a/core/src/test/resources/flows/valids/get-log-taskid.yaml
+++ b/core/src/test/resources/flows/valids/get-log-taskid.yaml
@@ -7,6 +7,10 @@ tasks:
   - type: io.kestra.plugin.core.debug.Echo
     id: task-2
     format: task 2
+  # as logs are async, to get your own logs reliably on tasks that run quickly, you must sleep a little
+  - type: io.kestra.plugin.core.flow.Sleep
+    id: sleep
+    duration: PT0.1S
   - type: io.kestra.plugin.core.log.Fetch
     id: get-log-task
     tasksId:
diff --git a/core/src/test/resources/flows/valids/get-log.yaml b/core/src/test/resources/flows/valids/get-log.yaml
index d1d64fc12..8ed4a1210 100644
--- a/core/src/test/resources/flows/valids/get-log.yaml
+++ b/core/src/test/resources/flows/valids/get-log.yaml
@@ -7,5 +7,9 @@ tasks:
   - type: io.kestra.plugin.core.debug.Echo
     id: task-2
     format: task 2
+  # as logs are async, to get your own logs reliably on tasks that run quickly, you must sleep a little
+  - type: io.kestra.plugin.core.flow.Sleep
+    id: sleep
+    duration: PT0.1S
   - type: io.kestra.plugin.core.log.Fetch
     id: get-log-task
\ No newline at end of file