From fd19334972317105ad934ecc2f3dbda2e02c3d92 Mon Sep 17 00:00:00 2001
From: Ludovic DEHON <tchiot.ludo@gmail.com>
Date: Fri, 16 Aug 2024 21:31:23 +0200
Subject: [PATCH] fix(webserver): create execution don't display error

close #4673
---
 .../controllers/api/ExecutionController.java  | 13 ++++++++++--
 .../api/ExecutionControllerTest.java          | 20 +++++++++++++++++++
 2 files changed, 31 insertions(+), 2 deletions(-)

diff --git a/webserver/src/main/java/io/kestra/webserver/controllers/api/ExecutionController.java b/webserver/src/main/java/io/kestra/webserver/controllers/api/ExecutionController.java
index 272bed22e0b..b631b5cda40 100644
--- a/webserver/src/main/java/io/kestra/webserver/controllers/api/ExecutionController.java
+++ b/webserver/src/main/java/io/kestra/webserver/controllers/api/ExecutionController.java
@@ -56,6 +56,7 @@
 import io.micronaut.scheduling.TaskExecutors;
 import io.micronaut.scheduling.annotation.ExecuteOn;
 import io.micronaut.validation.Validated;
+import io.netty.util.IllegalReferenceCountException;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.Parameter;
 import io.swagger.v3.oas.annotations.media.Content;
@@ -590,13 +591,21 @@ public Publisher<Execution> create(
                                 sink.success(item);
                             }
                         });
-                        sink.onDispose(() -> receive.run());
+
+                        sink.onDispose(receive::run);
                     }
                 } catch (IOException e) {
                     sink.error(new RuntimeException(e));
                 }
             })
-            .doOnError(t -> Flux.from(inputs).subscribeOn(Schedulers.boundedElastic()).blockLast()); // need to consume the inputs in case of error;
+            .doOnError(t -> {
+                // need to consume the inputs in case of error, that can failed, but we ignored
+                try {
+                    Flux.from(inputs).subscribeOn(Schedulers.boundedElastic()).blockLast();
+                } catch (IllegalReferenceCountException ignored) {
+
+                }
+            });
     }
 
     private List<Label> parseLabels(List<String> labels) {
diff --git a/webserver/src/test/java/io/kestra/webserver/controllers/api/ExecutionControllerTest.java b/webserver/src/test/java/io/kestra/webserver/controllers/api/ExecutionControllerTest.java
index e841da01e11..e2394aff995 100644
--- a/webserver/src/test/java/io/kestra/webserver/controllers/api/ExecutionControllerTest.java
+++ b/webserver/src/test/java/io/kestra/webserver/controllers/api/ExecutionControllerTest.java
@@ -130,6 +130,7 @@ private Execution triggerExecution(String namespace, String flowId, MultipartBod
             Execution.class
         );
     }
+
     private MultipartBody createInputsFlowBody() {
         // Trigger execution
         File applicationFile = new File(Objects.requireNonNull(
@@ -187,6 +188,25 @@ void trigger() {
         assertThat(notFound.getStatus(), is(HttpStatus.NOT_FOUND));
     }
 
+
+    @Test
+    void invalidInputs() {
+        MultipartBody.Builder builder = MultipartBody.builder()
+            .addPart("validatedString", "B-failed");
+        inputs.forEach((s, o) -> builder.addPart(s, o instanceof String ? (String) o : null));
+
+        HttpClientResponseException e = assertThrows(
+            HttpClientResponseException.class,
+            () -> triggerExecution(TESTS_FLOW_NS, "inputs", builder.build(), false)
+        );
+
+        String response = e.getResponse().getBody(String.class).orElseThrow();
+
+        assertThat(response, containsString("Invalid entity"));
+        assertThat(response, containsString("Invalid input for `validatedString`"));
+    }
+
+
     @Test
     void triggerAndWait() {
         Execution result = triggerInputsFlowExecution(true);