Remove vertx-kafka-client dependency from smallrye-reactive-messaging-kafka

Fixes quarkusio#22295

Author: ozangunalp

bom/application/pom.xml

@@ -5224,37 +5224,6 @@
                 <artifactId>client</artifactId>
                 <version>${scram-client.version}</version>
             </dependency>
-            <dependency>
-                <groupId>io.vertx</groupId>
-                <artifactId>vertx-kafka-client</artifactId>
-                <version>${vertx.version}</version>
-                <exclusions>
-                    <exclusion>
-                        <groupId>com.fasterxml.jackson.core</groupId>
-                        <artifactId>jackson-core</artifactId>
-                    </exclusion>
-                    <exclusion>
-                        <groupId>com.fasterxml.jackson.module</groupId>
-                        <artifactId>jackson-module-scala_2.12</artifactId>
-                    </exclusion>
-                    <exclusion>
-                        <groupId>org.scala-lang</groupId>
-                        <artifactId>scala-reflect</artifactId>
-                    </exclusion>
-                    <exclusion>
-                        <groupId>org.apache.kafka</groupId>
-                        <artifactId>kafka-clients</artifactId>
-                    </exclusion>
-                    <exclusion>
-                        <groupId>com.101tec</groupId>
-                        <artifactId>zkclient</artifactId>
-                    </exclusion>
-                    <exclusion>
-                        <groupId>org.slf4j</groupId>
-                        <artifactId>slf4j-log4j12</artifactId>
-                    </exclusion>
-                </exclusions>
-            </dependency>
             <dependency>
                 <groupId>com.cronutils</groupId>
                 <artifactId>cron-utils</artifactId>

extensions/smallrye-reactive-messaging-kafka/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/kafka/deployment/KafkaCodecProcessor.java

@@ -48,7 +48,6 @@
 import io.smallrye.mutiny.tuples.Functions.TriConsumer;
 import io.smallrye.reactive.messaging.kafka.KafkaConnector;
 import io.smallrye.reactive.messaging.kafka.commit.ProcessingState;
-import io.vertx.kafka.client.consumer.impl.KafkaReadStreamImpl;
 
 public class SmallRyeReactiveMessagingKafkaProcessor {
 
@@ -64,13 +63,6 @@ FeatureBuildItem feature() {
 
     @BuildStep
     public void build(BuildProducer<ReflectiveClassBuildItem> reflectiveClass) {
-        // Required for the throttled commit strategy
-        reflectiveClass.produce(
-                ReflectiveClassBuildItem.builder(KafkaReadStreamImpl.class)
-                        .fields(true)
-                        .methods(true)
-                        .constructors(true)
-                        .build());
         reflectiveClass.produce(new ReflectiveClassBuildItem(true, true, ProcessingState.class));
     }
 
@@ -680,9 +672,9 @@ private static boolean isRawMessage(Type type) {
             // Kafka types
             Map.entry(DotName.createSimple(org.apache.kafka.common.utils.Bytes.class.getName()), org.apache.kafka.common.serialization.BytesDeserializer.class.getName()),
             // Vert.x types
-            Map.entry(DotName.createSimple(io.vertx.core.buffer.Buffer.class.getName()),   io.vertx.kafka.client.serialization.BufferDeserializer.class.getName()),
-            Map.entry(DotName.createSimple(io.vertx.core.json.JsonObject.class.getName()), io.vertx.kafka.client.serialization.JsonObjectDeserializer.class.getName()),
-            Map.entry(DotName.createSimple(io.vertx.core.json.JsonArray.class.getName()),  io.vertx.kafka.client.serialization.JsonArrayDeserializer.class.getName())
+            Map.entry(DotName.createSimple(io.vertx.core.buffer.Buffer.class.getName()),   io.quarkus.kafka.client.serialization.BufferDeserializer.class.getName()),
+            Map.entry(DotName.createSimple(io.vertx.core.json.JsonObject.class.getName()), io.quarkus.kafka.client.serialization.JsonObjectDeserializer.class.getName()),
+            Map.entry(DotName.createSimple(io.vertx.core.json.JsonArray.class.getName()),  io.quarkus.kafka.client.serialization.JsonArrayDeserializer.class.getName())
     );
 
     private static final Map<DotName, String> KNOWN_SERIALIZERS = Map.ofEntries(
@@ -709,9 +701,9 @@ private static boolean isRawMessage(Type type) {
             // Kafka types
             Map.entry(DotName.createSimple(org.apache.kafka.common.utils.Bytes.class.getName()), org.apache.kafka.common.serialization.BytesSerializer.class.getName()),
             // Vert.x types
-            Map.entry(DotName.createSimple(io.vertx.core.buffer.Buffer.class.getName()),   io.vertx.kafka.client.serialization.BufferSerializer.class.getName()),
-            Map.entry(DotName.createSimple(io.vertx.core.json.JsonObject.class.getName()), io.vertx.kafka.client.serialization.JsonObjectSerializer.class.getName()),
-            Map.entry(DotName.createSimple(io.vertx.core.json.JsonArray.class.getName()),  io.vertx.kafka.client.serialization.JsonArraySerializer.class.getName())
+            Map.entry(DotName.createSimple(io.vertx.core.buffer.Buffer.class.getName()),   io.quarkus.kafka.client.serialization.BufferSerializer.class.getName()),
+            Map.entry(DotName.createSimple(io.vertx.core.json.JsonObject.class.getName()), io.quarkus.kafka.client.serialization.JsonObjectSerializer.class.getName()),
+            Map.entry(DotName.createSimple(io.vertx.core.json.JsonArray.class.getName()),  io.quarkus.kafka.client.serialization.JsonArraySerializer.class.getName())
     );
     // @formatter:on