fix(webserver): create execution don't display error

close #4673
kestra-io · Aug 22, 2024 · d9e18d5 · d9e18d5
1 parent b9aafb3
commit d9e18d5
Show file tree

Hide file tree

Showing 2 changed files with 31 additions and 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
@@ -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;
@@ -588,13 +589,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
@@ -131,6 +131,7 @@ private Execution triggerExecution(String namespace, String flowId, MultipartBod
             Execution.class
         );
     }
+
     private MultipartBody createInputsFlowBody() {
         // Trigger execution
         File applicationFile = new File(Objects.requireNonNull(
@@ -188,6 +189,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);