Repository: kestra-io/kestra (java)

Issue:
Creating flow outputs of type ARRAY fails and doesn't generate an error log 
### Describe the issue

This flow works:

```yaml
id: array_output
namespace: kestra.sandbox
description: Save and Execute the flow

tasks:
  - id: return
    type: io.kestra.plugin.core.output.OutputValues
    values:
      one: "1rstValue"
      two: "2ndValue"

outputs:
  - id: myout
    type: ARRAY
    value: '["{{ outputs.return.values.one }}", "{{ outputs.return.values.two }}"]'
```

but this one with "dash" syntax doesn't work (also tried with several variations of quotes)

```yaml
id: array_output
namespace: kestra.sandbox
description: Save and Execute the flow

tasks:
  - id: return
    type: io.kestra.plugin.core.output.OutputValues
    values:
      one: "1rstValue"
      two: "2ndValue"

outputs:
  - id: myout
    type: ARRAY
    value:
      - '"{{ outputs.return.values.one }}"'
      - '"{{ outputs.return.values.two }}"'
```

raising the following error:


ERROR jdbc-queue-Execution_0 flow.array_output [namespace: kestra.sandbox] [flow: array_output] [execution: 19JVhLwglJ0nprYzdkVrKq] Failed to render output values
jakarta.validation.ConstraintViolationException: myout: Invalid output for `myout`, Expected `ARRAY` but received `["1rstValue", "2ndValue"]` with errors:
```
class java.util.ArrayList cannot be cast to class java.lang.String (java.util.ArrayList and java.lang.String are in module java.base of loader 'bootstrap')
```, but received `["1rstValue", "2ndValue"]`

### Environment

- Kestra Version: 0.18.3
- Operating System (OS/Docker/Kubernetes): docker
- Java Version (if you don't run kestra in Docker):


Pull request description:
fix(core): handle array in outputs
Fixes #4769

I also added a log but you must switch to the "Temporal view" to be able to see it:
![image](https://github.com/user-attachments/assets/f24a79f3-d782-4ba1-abcc-5e675824710f)


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 73508341b..194141755 100644
--- a/core/src/main/java/io/kestra/core/runners/ExecutorService.java
+++ b/core/src/main/java/io/kestra/core/runners/ExecutorService.java
@@ -359,11 +359,11 @@ public class ExecutorService {
             .withState(executor.getExecution().guessFinalState(flow));
 
         if (flow.getOutputs() != null) {
+            RunContext runContext = runContextFactory.of(executor.getFlow(), executor.getExecution());
             try {
                 Map<String, Object> outputs = flow.getOutputs()
                     .stream()
                     .collect(HashMap::new, (map, entry) -> map.put(entry.getId(), entry.getValue()), Map::putAll);
-                RunContext runContext = runContextFactory.of(executor.getFlow(), executor.getExecution());
                 outputs = runContext.render(outputs);
                 outputs = flowInputOutput.typedOutputs(flow, executor.getExecution(), outputs);
                 newExecution = newExecution.withOutputs(outputs);
@@ -375,8 +375,8 @@ public class ExecutorService {
                     "Failed to render output values",
                     e
                 );
-                newExecution = newExecution
-                    .withState(State.Type.FAILED);
+                runContext.logger().error("Failed to render output values: {}", e.getMessage(), e);
+                newExecution = newExecution.withState(State.Type.FAILED);
             }
         }
 
diff --git a/core/src/main/java/io/kestra/core/runners/FlowInputOutput.java b/core/src/main/java/io/kestra/core/runners/FlowInputOutput.java
index 70ee4fa2e..7577d1241 100644
--- a/core/src/main/java/io/kestra/core/runners/FlowInputOutput.java
+++ b/core/src/main/java/io/kestra/core/runners/FlowInputOutput.java
@@ -293,10 +293,16 @@ public class FlowInputOutput {
                     }
                 }
                 case ARRAY, MULTISELECT -> {
+                    List<?> asList;
+                    if (current instanceof List<?> list) {
+                        asList = list;
+                    } else {
+                        asList = JacksonMapper.toList(((String) current));
+                    }
+
                     if (elementType != null) {
                         // recursively parse the elements only once
-                        yield JacksonMapper.toList(((String) current))
-                            .stream()
+                        yield asList.stream()
                             .map(throwFunction(element -> {
                                 try {
                                     return parseType(execution, elementType, id, null, element);
@@ -306,7 +312,7 @@ public class FlowInputOutput {
                             }))
                             .toList();
                     } else {
-                        yield JacksonMapper.toList(((String) current));
+                        yield asList;
                     }
                 }
             };


Test patch:
diff --git a/core/src/test/java/io/kestra/plugin/core/flow/FlowOutputTest.java b/core/src/test/java/io/kestra/plugin/core/flow/FlowOutputTest.java
index 0ff39b17c..62416ecbf 100644
--- a/core/src/test/java/io/kestra/plugin/core/flow/FlowOutputTest.java
+++ b/core/src/test/java/io/kestra/plugin/core/flow/FlowOutputTest.java
@@ -5,25 +5,33 @@ import io.kestra.core.models.flows.State;
 import io.kestra.core.runners.AbstractMemoryRunnerTest;
 import org.junit.jupiter.api.Test;
 
+import java.util.List;
 import java.util.concurrent.TimeoutException;
 
 import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.aMapWithSize;
-import static org.hamcrest.Matchers.is;
-import static org.hamcrest.Matchers.nullValue;
+import static org.hamcrest.Matchers.*;
 
 class FlowOutputTest extends AbstractMemoryRunnerTest {
 
     static final String NAMESPACE = "io.kestra.tests";
 
     @Test
-    void shouldGetSuccessExecutionForFlowFlowWithOutputs() throws TimeoutException {
+    void shouldGetSuccessExecutionForFlowWithOutputs() throws TimeoutException {
         Execution execution = runnerUtils.runOne(null, NAMESPACE, "flow-with-outputs", null, null);
         assertThat(execution.getOutputs(), aMapWithSize(1));
         assertThat(execution.getOutputs().get("key"), is("{\"value\":\"flow-with-outputs\"}"));
         assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
     }
 
+    @Test
+    @SuppressWarnings("unchecked")
+    void shouldGetSuccessExecutionForFlowWithArrayOutputs() throws TimeoutException {
+        Execution execution = runnerUtils.runOne(null, NAMESPACE, "flow-with-array-outputs", null, null);
+        assertThat(execution.getOutputs(), aMapWithSize(1));
+        assertThat((List<String>) execution.getOutputs().get("myout"), hasItems("1rstValue", "2ndValue"));
+        assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
+    }
+
     @Test
     void shouldGetFailExecutionForFlowWithInvalidOutputs() throws TimeoutException {
         Execution execution = runnerUtils.runOne(null, NAMESPACE, "flow-with-outputs-failed", null, null);
diff --git a/core/src/test/resources/flows/valids/flow-with-array-outputs.yml b/core/src/test/resources/flows/valids/flow-with-array-outputs.yml
new file mode 100644
index 000000000..c7c738912
--- /dev/null
+++ b/core/src/test/resources/flows/valids/flow-with-array-outputs.yml
@@ -0,0 +1,16 @@
+id: flow-with-array-outputs
+namespace: io.kestra.tests
+
+tasks:
+  - id: return
+    type: io.kestra.plugin.core.output.OutputValues
+    values:
+      one: "1rstValue"
+      two: "2ndValue"
+
+outputs:
+  - id: myout
+    type: ARRAY
+    value:
+      - "{{ outputs.return.values.one }}"
+      - "{{ outputs.return.values.two }}"
\ No newline at end of file