Repository: quarkusio/quarkus
Issue: Eventbus, No message codec for type when using headers

### Describe the bug

When a event class is only consumed by `@ConsumeEvent` methods with signature 
```java
@ConsumeEvent(value = "address")
public void listen(MultiMap headers, Body body) {
        ... 
}
```
no message codec is regisered for this class. 
Resulting in 
```shell
java.lang.IllegalArgumentException: No message codec for type: class org.acme.Hello
	at io.vertx.core.eventbus.impl.CodecManager.lookupCodec(CodecManager.java:119)
	at io.vertx.core.eventbus.impl.EventBusImpl.createMessage(EventBusImpl.java:254)
	at io.vertx.core.eventbus.impl.EventBusImpl.publish(EventBusImpl.java:164)
	at io.vertx.core.eventbus.impl.EventBusImpl.publish(EventBusImpl.java:159)
	at org.acme.GreetingResource.hello(GreetingResource.java:23)
	at org.acme.GreetingResource$quarkusrestinvoker$hello_e747664148511e1e5212d3e0f4b40d45c56ab8a1.invoke(Unknown Source)
	at org.jboss.resteasy.reactive.server.handlers.InvocationHandler.handle(InvocationHandler.java:29)
	at io.quarkus.resteasy.reactive.server.runtime.QuarkusResteasyReactiveRequestContext.invokeHandler(QuarkusResteasyReactiveRequestContext.java:114)
	at org.jboss.resteasy.reactive.common.core.AbstractResteasyReactiveContext.run(AbstractResteasyReactiveContext.java:145)
	at io.quarkus.vertx.core.runtime.VertxCoreRecorder$14.runWith(VertxCoreRecorder.java:576)
	at org.jboss.threads.EnhancedQueueExecutor$Task.run(EnhancedQueueExecutor.java:2449)
	at org.jboss.threads.EnhancedQueueExecutor$ThreadBody.run(EnhancedQueueExecutor.java:1478)
	at org.jboss.threads.DelegatingRunnable.run(DelegatingRunnable.java:29)
	at org.jboss.threads.ThreadLocalResettingRunnable.run(ThreadLocalResettingRunnable.java:29)
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	at java.base/java.lang.Thread.run(Thread.java:833)
```
Adding another method without headers will register a codec.

### Expected behavior

Message codecs are created on startup regardless of whether the event consumers include headers.

### Actual behavior

No message codecs are created, resulting in 
```shell
java.lang.IllegalArgumentException: No message codec for type: class org.acme.Hello
	at io.vertx.core.eventbus.impl.CodecManager.lookupCodec(CodecManager.java:119)
	at io.vertx.core.eventbus.impl.EventBusImpl.createMessage(EventBusImpl.java:254)
	at io.vertx.core.eventbus.impl.EventBusImpl.publish(EventBusImpl.java:164)
	at io.vertx.core.eventbus.impl.EventBusImpl.publish(EventBusImpl.java:159)
	at org.acme.GreetingResource.hello(GreetingResource.java:23)
	at org.acme.GreetingResource$quarkusrestinvoker$hello_e747664148511e1e5212d3e0f4b40d45c56ab8a1.invoke(Unknown Source)
	at org.jboss.resteasy.reactive.server.handlers.InvocationHandler.handle(InvocationHandler.java:29)
	at io.quarkus.resteasy.reactive.server.runtime.QuarkusResteasyReactiveRequestContext.invokeHandler(QuarkusResteasyReactiveRequestContext.java:114)
	at org.jboss.resteasy.reactive.common.core.AbstractResteasyReactiveContext.run(AbstractResteasyReactiveContext.java:145)
	at io.quarkus.vertx.core.runtime.VertxCoreRecorder$14.runWith(VertxCoreRecorder.java:576)
	at org.jboss.threads.EnhancedQueueExecutor$Task.run(EnhancedQueueExecutor.java:2449)
	at org.jboss.threads.EnhancedQueueExecutor$ThreadBody.run(EnhancedQueueExecutor.java:1478)
	at org.jboss.threads.DelegatingRunnable.run(DelegatingRunnable.java:29)
	at org.jboss.threads.ThreadLocalResettingRunnable.run(ThreadLocalResettingRunnable.java:29)
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	at java.base/java.lang.Thread.run(Thread.java:833)
```

### How to Reproduce?

https://github.com/computerlove/quarkus-evenbus-headers-reproducer
Start with `mvn quarkus:dev` and GET http://localhost:8080/hello
Exception will be printed in terminal. 
Commenting in the extra listener, `listenNoHeaders`, in `Listener.java` will result in the expected behaviour.

### Output of `uname -a` or `ver`

_No response_

### Output of `java -version`

java 17

### GraalVM version (if different from Java)

_No response_

### Quarkus version or git rev

2.15.3

### Build tool (ie. output of `mvnw --version` or `gradlew --version`)

_No response_

### Additional information

_No response_

Changed files:
integration-tests/vertx/src/main/java/io/quarkus/it/vertx/EventBusSender.java
integration-tests/vertx/src/main/java/io/quarkus/it/vertx/EventBusConsumer.java
integration-tests/vertx/src/test/java/io/quarkus/it/vertx/EventBusTest.java
extensions/vertx/deployment/src/test/java/io/quarkus/vertx/CodecRegistrationTest.java
extensions/vertx/deployment/src/main/java/io/quarkus/vertx/deployment/EventBusCodecProcessor.java

diff --git a/extensions/vertx/deployment/src/main/java/io/quarkus/vertx/deployment/EventBusCodecProcessor.java b/extensions/vertx/deployment/src/main/java/io/quarkus/vertx/deployment/EventBusCodecProcessor.java
index 27ae071ec82..6cc48b1848f 100644
--- a/extensions/vertx/deployment/src/main/java/io/quarkus/vertx/deployment/EventBusCodecProcessor.java
+++ b/extensions/vertx/deployment/src/main/java/io/quarkus/vertx/deployment/EventBusCodecProcessor.java
@@ -168,7 +168,12 @@ private static Type extractPayloadTypeFromParameter(MethodInfo method) {
         if (parameters.isEmpty()) {
             return null;
         }
-        Type param = method.parameterType(0);
+        /*
+         * VertxProcessor.collectEventConsumers makes sure that only methods with either just the message object,
+         * or headers as first argument then message object are allowed.
+         */
+        int messageIndex = parameters.size() == 1 ? 0 : 1;
+        Type param = method.parameterType(messageIndex);
         if (param.kind() == Type.Kind.CLASS) {
             return param;
         } else if (param.kind() == Type.Kind.PARAMETERIZED_TYPE) {
diff --git a/extensions/vertx/deployment/src/test/java/io/quarkus/vertx/CodecRegistrationTest.java b/extensions/vertx/deployment/src/test/java/io/quarkus/vertx/CodecRegistrationTest.java
index 58bfeb0b163..046e2468258 100644
--- a/extensions/vertx/deployment/src/test/java/io/quarkus/vertx/CodecRegistrationTest.java
+++ b/extensions/vertx/deployment/src/test/java/io/quarkus/vertx/CodecRegistrationTest.java
@@ -8,6 +8,7 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
 import javax.inject.Inject;
@@ -18,6 +19,7 @@
 import org.junit.jupiter.api.extension.RegisterExtension;
 
 import io.quarkus.test.QuarkusUnitTest;
+import io.vertx.core.MultiMap;
 import io.vertx.mutiny.core.Vertx;
 import io.vertx.mutiny.core.eventbus.Message;
 
@@ -83,6 +85,8 @@ public void testWithPrimitiveTypesAndCompletionStage() {
 
     @Test
     public void testCodecRegistrationBasedOnParameterType() {
+        assertThat(bean.getSink().size()).isEqualTo(0);
+
         String address = "address-5";
         vertx.eventBus().send(address, new CustomType1("foo"));
         vertx.eventBus().send(address, new CustomType1("bar"));
@@ -104,6 +108,20 @@ public void testCodecRegistrationBasedOnParameterType() {
         set = bean.getSink().stream().map(x -> (CustomType1) x).map(CustomType1::getName)
                 .collect(Collectors.toSet());
         assertThat(set).contains("foo-x", "bar-x", "baz-x");
+        bean.getSink().clear();
+    }
+
+    @Test
+    public void testCodecRegistrationBasedOnHeadersAndParameterType() {
+        assertThat(bean.getSink().size()).isEqualTo(0);
+
+        vertx.eventBus().send("address-9", new CustomType5("foo-x"));
+
+        await().timeout(5, TimeUnit.SECONDS).until(() -> bean.getSink().size() == 1);
+        Set<String> set = bean.getSink().stream().map(x -> (CustomType5) x).map(CustomType5::getName)
+                .collect(Collectors.toSet());
+        assertThat(set).contains("foo-x");
+        bean.getSink().clear();
     }
 
     @Test
@@ -199,6 +217,11 @@ CompletionStage<CustomType4> codecRegistrationBasedReturnTypeAndCS(String n) {
             return CompletableFuture.completedFuture(new CustomType4(n));
         }
 
+        @ConsumeEvent("address-9")
+        void codecRegistrationBasedOnHeadersParam(MultiMap headers, CustomType5 ct) {
+            sink.add(ct);
+        }
+
         public List<Object> getSink() {
             return sink;
         }
@@ -259,4 +282,16 @@ public String getName() {
             return name;
         }
     }
+
+    static class CustomType5 {
+        private final String name;
+
+        CustomType5(String name) {
+            this.name = name;
+        }
+
+        public String getName() {
+            return name;
+        }
+    }
 }
diff --git a/integration-tests/vertx/src/main/java/io/quarkus/it/vertx/EventBusConsumer.java b/integration-tests/vertx/src/main/java/io/quarkus/it/vertx/EventBusConsumer.java
index 419edb41ee9..1cfb1af7920 100644
--- a/integration-tests/vertx/src/main/java/io/quarkus/it/vertx/EventBusConsumer.java
+++ b/integration-tests/vertx/src/main/java/io/quarkus/it/vertx/EventBusConsumer.java
@@ -3,6 +3,7 @@
 import javax.enterprise.context.ApplicationScoped;
 
 import io.quarkus.vertx.ConsumeEvent;
+import io.vertx.core.MultiMap;
 
 @ApplicationScoped
 public class EventBusConsumer {
@@ -17,4 +18,10 @@ public String name(String name) {
         return "Hello " + name;
     }
 
+    @ConsumeEvent("person-headers")
+    public String personWithHeader(MultiMap headers, Person person) {
+        String s = "Hello " + person.getFirstName() + " " + person.getLastName() + ", " + headers;
+        return s;
+    }
+
 }
diff --git a/integration-tests/vertx/src/main/java/io/quarkus/it/vertx/EventBusSender.java b/integration-tests/vertx/src/main/java/io/quarkus/it/vertx/EventBusSender.java
index d623a40c3bf..46b21d3df93 100644
--- a/integration-tests/vertx/src/main/java/io/quarkus/it/vertx/EventBusSender.java
+++ b/integration-tests/vertx/src/main/java/io/quarkus/it/vertx/EventBusSender.java
@@ -3,8 +3,10 @@
 import javax.inject.Inject;
 import javax.ws.rs.POST;
 import javax.ws.rs.Path;
+import javax.ws.rs.Produces;
 
 import io.smallrye.mutiny.Uni;
+import io.vertx.core.eventbus.DeliveryOptions;
 import io.vertx.core.json.JsonObject;
 import io.vertx.mutiny.core.eventbus.EventBus;
 import io.vertx.mutiny.core.eventbus.Message;
@@ -22,6 +24,17 @@ public Uni<String> helloToPerson(JsonObject json) {
                 .onItem().transform(Message::body);
     }
 
+    @POST
+    @Path("/person2")
+    @Produces("text/plain")
+    public Uni<String> helloToPersonWithHeaders(JsonObject json) {
+        return bus.<String> request(
+                "person-headers",
+                new Person(json.getString("firstName"), json.getString("lastName")),
+                new DeliveryOptions().addHeader("header", "headerValue"))
+                .onItem().transform(Message::body);
+    }
+
     @POST
     @Path("/pet")
     public Uni<String> helloToPet(JsonObject json) {
diff --git a/integration-tests/vertx/src/test/java/io/quarkus/it/vertx/EventBusTest.java b/integration-tests/vertx/src/test/java/io/quarkus/it/vertx/EventBusTest.java
index 31ada479f39..be8f78e784d 100644
--- a/integration-tests/vertx/src/test/java/io/quarkus/it/vertx/EventBusTest.java
+++ b/integration-tests/vertx/src/test/java/io/quarkus/it/vertx/EventBusTest.java
@@ -20,6 +20,19 @@ public void testEventBusWithString() {
                 .then().statusCode(200).body(equalTo("Hello Bob Morane"));
     }
 
+    @Test
+    public void testEventBusWithObjectAndHeader() {
+        String body = new JsonObject()
+                .put("firstName", "Bob")
+                .put("lastName", "Morane")
+                .toString();
+        given().contentType(ContentType.JSON).body(body)
+                .post("/vertx-test/event-bus/person2")
+                .then().statusCode(200)
+                // For some reason Multimap.toString() has \\n at the end.
+                .body(equalTo("Hello Bob Morane, header=headerValue\\n"));
+    }
+
     @Test
     public void testEventBusWithPet() {
         String body = new JsonObject().put("name", "Neo").put("kind", "rabbit").toString();