From 383431258c961bea32814ad31aa521cade883549 Mon Sep 17 00:00:00 2001
From: Jaikiran Pai <jaikiran.pai@gmail.com>
Date: Tue, 6 Apr 2021 09:25:06 +0530
Subject: [PATCH] Don't let potential failures in message consumers cause dev
 mode shutdown to be blocked permanently

---
 .../io/quarkus/vertx/runtime/VertxRecorder.java   | 15 +++++++++++----
 1 file changed, 11 insertions(+), 4 deletions(-)

diff --git a/extensions/vertx/runtime/src/main/java/io/quarkus/vertx/runtime/VertxRecorder.java b/extensions/vertx/runtime/src/main/java/io/quarkus/vertx/runtime/VertxRecorder.java
index 54de1a6856f8c..69c8395a18633 100644
--- a/extensions/vertx/runtime/src/main/java/io/quarkus/vertx/runtime/VertxRecorder.java
+++ b/extensions/vertx/runtime/src/main/java/io/quarkus/vertx/runtime/VertxRecorder.java
@@ -70,6 +70,7 @@ void registerMessageConsumers(Map<String, ConsumeEvent> messageConsumerConfigura
         if (!messageConsumerConfigurations.isEmpty()) {
             EventBus eventBus = vertx.eventBus();
             CountDownLatch latch = new CountDownLatch(messageConsumerConfigurations.size());
+            final List<Throwable> registrationFailures = new ArrayList();
             for (Entry<String, ConsumeEvent> entry : messageConsumerConfigurations.entrySet()) {
                 EventConsumerInvoker invoker = createInvoker(entry.getKey());
                 String address = entry.getValue().value();
@@ -117,8 +118,9 @@ public void handle(Promise<Object> event) {
 
                     @Override
                     public void handle(AsyncResult<Void> ar) {
-                        if (ar.succeeded()) {
-                            latch.countDown();
+                        latch.countDown();
+                        if (ar.failed()) {
+                            registrationFailures.add(ar.cause());
                         }
                     }
                 });
@@ -130,6 +132,10 @@ public void handle(AsyncResult<Void> ar) {
                 Thread.currentThread().interrupt();
                 throw new IllegalStateException("Unable to register all message consumer methods", e);
             }
+            if (!registrationFailures.isEmpty()) {
+                // just log/raise the first failure
+                throw new RuntimeException("Registration of one or more message consumers failed", registrationFailures.get(0));
+            }
         }
     }
 
@@ -147,8 +153,9 @@ void unregisterMessageConsumers() {
         CountDownLatch latch = new CountDownLatch(messageConsumers.size());
         for (MessageConsumer<?> messageConsumer : messageConsumers) {
             messageConsumer.unregister(ar -> {
-                if (ar.succeeded()) {
-                    latch.countDown();
+                latch.countDown();
+                if (ar.failed()) {
+                    LOGGER.warn("Message consumer unregistration failed", ar.cause());
                 }
             });
         }