Merge pull request #16253 from jaikiran/qk-16230

Don't let potential failures in message consumers cause dev mode shutdown to be blocked permanently
quarkusio · Apr 8, 2021 · 6a6f82b · 6a6f82b
2 parents b6ddd49 + 3834312
commit 6a6f82b
Showing 1 changed file with 11 additions and 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
@@ -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());
                 }
             });
         }