Repository: quarkusio/quarkus
Issue: Blocking @ConsumeEvent only uses single worker thread since 1.11.X

### Description
Greetings,

while upgrading our Quarkus services from 1.10.4 to 1.11.3 we noticed that the behaviour of the Vert.x event bus significally changed. The event bus does not utilize more than one thread from the worker thread pool and the execution of multiple events is therefore not concurrent as it was prior to 1.11.X.

I could not find any hints in the changelogs or the migration guide why the behaviour changed.

If this new behaviour is indeed expectec, how can we handle the conccurrent execution of events if we do not want to use reactive style code?

### Expected behavior
ConsumeEvent with `blocking=true` should utilize all available worker pool threads and execute events concrrently.

### Actual behavior
The execution of events seems to only use a single worker thread and does not execute events concurrently.

### Example Code
Bean with some long running job:
```
@Dependent
public class SomeJobClass {

    private static final Logger LOG = Logger.getLogger(SomeJobClass.class);

    public void longRunningJob(String test) {
        LOG.info("Executing");
        try {
            Thread.sleep(1000l);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        LOG.info("finished");
    }
}
```

The class containing the consumer:
```
@ApplicationScoped
public class ConsumerClass {

    @Inject
    SomeJobClass importer;

    @ConsumeEvent(value = "test", blocking = true)
    void test(String event) {
        importer.longRunningJob(event);
    }
}
```

Send 2 events:
```
@Test
@DisplayName("Some Test")
void testConcurrentEvents() throws InterruptedException {
    eventBus.send("test", "1");
    eventBus.send("test", "1");

    Thread.sleep(3000l);
    }
```


Log output with Quarkus 1.11.2.FINAL:
```
﻿﻿2021-02-26 13:53:29,994 INFO  [foo] (vert.x-worker-thread-0) Executing
2021-02-26 13:53:30,997 INFO  [foo] (vert.x-worker-thread-0) finished
2021-02-26 13:53:30,998 INFO  [foo] (vert.x-worker-thread-0) Executing
2021-02-26 13:53:32,002 INFO  [foo] (vert.x-worker-thread-0) finished
```
Log output with Quarkus 1.10.5.FINAL:
```
2021-02-26 14:02:19,345 INFO  [foo] (vert.x-worker-thread-0) Executing
2021-02-26 14:02:19,345 INFO  [foo] (vert.x-worker-thread-1) Executing
2021-02-26 14:02:20,349 INFO  [foo] (vert.x-worker-thread-0) finished
2021-02-26 14:02:20,349 INFO  [foo] (vert.x-worker-thread-1) finished
```
The logs show that the execution with version 1.11.2.FINAL was not concurrent and is using only `vert.x-worker-thread-0`.



Changed files:
extensions/vertx/runtime/src/main/java/io/quarkus/vertx/runtime/VertxRecorder.java
extensions/vertx/runtime/src/main/java/io/quarkus/vertx/runtime/EventConsumerInvoker.java
extensions/vertx/deployment/src/main/java/io/quarkus/vertx/deployment/EventBusConsumer.java
extensions/vertx/runtime/src/main/java/io/quarkus/vertx/ConsumeEvent.java

diff --git a/extensions/vertx/deployment/src/main/java/io/quarkus/vertx/deployment/EventBusConsumer.java b/extensions/vertx/deployment/src/main/java/io/quarkus/vertx/deployment/EventBusConsumer.java
index 57b90701781..b5d97cbe628 100644
--- a/extensions/vertx/deployment/src/main/java/io/quarkus/vertx/deployment/EventBusConsumer.java
+++ b/extensions/vertx/deployment/src/main/java/io/quarkus/vertx/deployment/EventBusConsumer.java
@@ -109,6 +109,13 @@ static String generateInvoker(BeanInfo bean, MethodInfo method,
             isBlocking.returnValue(isBlocking.load(true));
         }
 
+        AnnotationValue orderedValue = consumeEvent.value("ordered");
+        boolean ordered = orderedValue != null && orderedValue.asBoolean();
+        if (ordered) {
+            MethodCreator isOrdered = invokerCreator.getMethodCreator("isOrdered", boolean.class);
+            isOrdered.returnValue(isOrdered.load(true));
+        }
+
         implementConstructor(bean, invokerCreator, beanField, containerField);
         implementInvoke(bean, method, invokerCreator, beanField.getFieldDescriptor(), containerField.getFieldDescriptor());
 
diff --git a/extensions/vertx/runtime/src/main/java/io/quarkus/vertx/ConsumeEvent.java b/extensions/vertx/runtime/src/main/java/io/quarkus/vertx/ConsumeEvent.java
index 0020c381a43..e9a870edb29 100644
--- a/extensions/vertx/runtime/src/main/java/io/quarkus/vertx/ConsumeEvent.java
+++ b/extensions/vertx/runtime/src/main/java/io/quarkus/vertx/ConsumeEvent.java
@@ -83,12 +83,19 @@
      */
     boolean blocking() default false;
 
+    /**
+     * @return {@code true} if the <em>blocking</em> consumption of the event must be ordered, meaning that the method
+     *         won't be called concurrently. Instead it serializes all the invocations based on the event order.
+     *         {@code ordered} must be used in conjunction with {@code blocking=true} or {@code @Blocking}.
+     * @see io.vertx.core.Vertx#executeBlocking(io.vertx.core.Handler, boolean, io.vertx.core.Handler)
+     */
+    boolean ordered() default false;
+
     /**
      * 
      * @return {@code null} if it should use a default MessageCodec
      * @see io.quarkus.vertx.LocalEventBusCodec
      */
-    @SuppressWarnings("rawtypes")
     Class<? extends MessageCodec> codec() default LocalEventBusCodec.class;
 
 }
diff --git a/extensions/vertx/runtime/src/main/java/io/quarkus/vertx/runtime/EventConsumerInvoker.java b/extensions/vertx/runtime/src/main/java/io/quarkus/vertx/runtime/EventConsumerInvoker.java
index 6320aa80243..38acccc6168 100644
--- a/extensions/vertx/runtime/src/main/java/io/quarkus/vertx/runtime/EventConsumerInvoker.java
+++ b/extensions/vertx/runtime/src/main/java/io/quarkus/vertx/runtime/EventConsumerInvoker.java
@@ -18,6 +18,10 @@ public boolean isBlocking() {
         return false;
     }
 
+    public boolean isOrdered() {
+        return false;
+    }
+
     public void invoke(Message<Object> message) throws Exception {
         ManagedContext requestContext = Arc.container().requestContext();
         if (requestContext.isActive()) {
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 b340eabd863..54de1a6856f 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
@@ -98,7 +98,7 @@ public void handle(Promise<Object> event) {
                                     }
                                     event.complete();
                                 }
-                            }, null);
+                            }, invoker.isOrdered(), null);
                         } else {
                             try {
                                 invoker.invoke(m);