Fix exception handling when a message consumer is blocking

extensions/vertx/deployment/src/main/java/io/quarkus/vertx/deployment/EventBusConsumer.java

@@ -1,5 +1,6 @@
 package io.quarkus.vertx.deployment;
 
+import static io.quarkus.vertx.ConsumeEvent.FAILURE_CODE;
 import static io.quarkus.vertx.deployment.VertxConstants.*;
 
 import java.lang.annotation.Annotation;
@@ -85,6 +86,8 @@ class EventBusConsumer {
             .ofMethod(Uni.class, "subscribeAsCompletionStage", CompletableFuture.class);
     protected static final MethodDescriptor THROWABLE_GET_MESSAGE = MethodDescriptor
             .ofMethod(Throwable.class, "getMessage", String.class);
+    protected static final MethodDescriptor THROWABLE_TO_STRING = MethodDescriptor
+            .ofMethod(Throwable.class, "toString", String.class);
 
     static String generateInvoker(BeanInfo bean, MethodInfo method,
             AnnotationInstance consumeEvent,
@@ -130,8 +133,20 @@ static String generateInvoker(BeanInfo bean, MethodInfo method,
             TryBlock tryBlock = funcBytecode.tryBlock();
             invoke(bean, method, messageHandle, tryBlock);
             tryBlock.invokeInterfaceMethod(FUTURE_COMPLETE, funcBytecode.getMethodParam(0), tryBlock.loadNull());
+
             CatchBlockCreator catchBlock = tryBlock.addCatch(Exception.class);
-            catchBlock.invokeInterfaceMethod(FUTURE_FAIL, funcBytecode.getMethodParam(0), catchBlock.getCaughtException());
+            // Need to reply with the caught exception - using Throwable.toString on purpose to get the class name.
+            ResultHandle failureMessage = catchBlock
+                    .invokeVirtualMethod(THROWABLE_TO_STRING, catchBlock.getCaughtException());
+            ResultHandle failureStatus = catchBlock.load(FAILURE_CODE);
+            catchBlock.invokeInterfaceMethod(
+                    MESSAGE_FAIL,
+                    messageHandle,
+                    failureStatus,
+                    failureMessage);
+            // Completing successfully, the failure has been sent to the sender.
+            catchBlock.invokeInterfaceMethod(FUTURE_COMPLETE, funcBytecode.getMethodParam(0), catchBlock.loadNull());
+
             funcBytecode.returnValue(null);
 
             invoke.invokeInterfaceMethod(VERTX_EXECUTE_BLOCKING, vertxHandle, func.getInstance(), invoke.load(false),

extensions/vertx/deployment/src/test/java/io/quarkus/vertx/deployment/MessageConsumerFailureTest.java

@@ -44,6 +44,13 @@ public void testFailure() throws InterruptedException {
         verifyFailure("foo-completion-stage-failure", "boom", true);
         verifyFailure("foo-uni", "java.lang.NullPointerException: Something is null", false);
         verifyFailure("foo-uni-failure", "boom", true);
+
+        verifyFailure("foo-blocking", "java.lang.IllegalStateException: Red is dead", false);
+        verifyFailure("foo-message-blocking", "java.lang.NullPointerException", false);
+        verifyFailure("foo-completion-stage-blocking", "java.lang.NullPointerException: Something is null", false);
+        verifyFailure("foo-completion-stage-failure-blocking", "boom", true);
+        verifyFailure("foo-uni-blocking", "java.lang.NullPointerException: Something is null", false);
+        verifyFailure("foo-uni-failure-blocking", "boom", true);
     }
 
     void verifyFailure(String address, String expectedMessage, boolean explicit) throws InterruptedException {
@@ -104,6 +111,37 @@ Uni<String> failedUni(String message) {
             return Uni.createFrom().failure(new IOException("boom"));
         }
 
+        @ConsumeEvent(value = "foo-blocking", blocking = true)
+        String failBlocking(String message) {
+            throw new IllegalStateException("Red is dead");
+        }
+
+        @ConsumeEvent(value = "foo-message-blocking", blocking = true)
+        void failMessageBlocking(Message<String> message) {
+            throw new NullPointerException();
+        }
+
+        @ConsumeEvent(value = "foo-completion-stage-blocking", blocking = true)
+        CompletionStage<String> failCompletionStageBlocking(String message) {
+            throw new NullPointerException("Something is null");
+        }
+
+        @ConsumeEvent(value = "foo-completion-stage-failure-blocking", blocking = true)
+        CompletionStage<String> failedCompletionStageBlocking(String message) {
+            CompletableFuture<String> future = new CompletableFuture<>();
+            future.completeExceptionally(new IOException("boom"));
+            return future;
+        }
+
+        @ConsumeEvent(value = "foo-uni-blocking", blocking = true)
+        Uni<String> failUniBlocking(String message) {
+            throw new NullPointerException("Something is null");
+        }
+
+        @ConsumeEvent(value = "foo-uni-failure-blocking", blocking = true)
+        Uni<String> failedUniBlocking(String message) {
+            return Uni.createFrom().failure(new IOException("boom"));
+        }
     }
 
 }