Handle subclasses for Kafka Serde autodetection

Added provided serde classes to the jandex index
quarkusio · Sep 17, 2024 · 280f66d · 280f66d
1 parent c3e7ee5
commit 280f66d
Show file tree

Hide file tree

Showing 4 changed files with 88 additions and 14 deletions.
diff --git a/...ka-client/deployment/src/main/java/io/quarkus/kafka/client/deployment/KafkaProcessor.java b/...ka-client/deployment/src/main/java/io/quarkus/kafka/client/deployment/KafkaProcessor.java
@@ -197,8 +197,21 @@ void addSaslProvidersToNativeImage(BuildProducer<NativeImageSecurityProviderBuil
 
     @BuildStep
     void contributeClassesToIndex(BuildProducer<AdditionalIndexedClassesBuildItem> additionalIndexedClasses,
-            BuildProducer<IndexDependencyBuildItem> indexDependency) {
+            BuildProducer<IndexDependencyBuildItem> indexDependency, Capabilities capabilities) {
         indexDependency.produce(new IndexDependencyBuildItem("org.apache.kafka", "kafka-clients"));
+        additionalIndexedClasses.produce(new AdditionalIndexedClassesBuildItem(
+                JsonObjectSerializer.class.getName(),
+                JsonObjectDeserializer.class.getName(),
+                JsonArraySerializer.class.getName(),
+                JsonArrayDeserializer.class.getName(),
+                BufferSerializer.class.getName(),
+                BufferDeserializer.class.getName(),
+                ObjectMapperSerializer.class.getName(),
+                ObjectMapperDeserializer.class.getName()));
+        if (capabilities.isPresent(Capability.JSONB)) {
+            additionalIndexedClasses.produce(new AdditionalIndexedClassesBuildItem(
+                    JsonbSerializer.class.getName(), JsonbDeserializer.class.getName()));
+        }
     }
 
     @BuildStep

diff --git a/...va/io/quarkus/smallrye/reactivemessaging/kafka/deployment/DefaultSerdeDiscoveryState.java b/...va/io/quarkus/smallrye/reactivemessaging/kafka/deployment/DefaultSerdeDiscoveryState.java
@@ -2,6 +2,7 @@
 
 import static io.quarkus.smallrye.reactivemessaging.kafka.deployment.SmallRyeReactiveMessagingKafkaProcessor.getChannelPropertyKey;
 
+import java.util.Comparator;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -17,8 +18,8 @@
 import org.jboss.jandex.ClassInfo;
 import org.jboss.jandex.DotName;
 import org.jboss.jandex.IndexView;
-import org.jboss.jandex.Type;
 
+import io.quarkus.deployment.util.JandexUtil;
 import io.quarkus.smallrye.reactivemessaging.deployment.items.ChannelDirection;
 import io.quarkus.smallrye.reactivemessaging.deployment.items.ConnectorManagedChannelBuildItem;
 import io.smallrye.reactive.messaging.kafka.KafkaConnector;
@@ -184,24 +185,20 @@ boolean hasJsonb() {
     }
 
     ClassInfo getSubclassOfWithTypeArgument(DotName superclass, DotName expectedTypeArgument) {
-        return index.getKnownDirectSubclasses(superclass)
+        return index.getAllKnownSubclasses(superclass)
                 .stream()
-                .filter(it -> it.superClassType().kind() == Type.Kind.PARAMETERIZED_TYPE
-                        && it.superClassType().asParameterizedType().arguments().size() == 1
-                        && it.superClassType().asParameterizedType().arguments().get(0).name().equals(expectedTypeArgument))
-                .findAny()
+                .filter(ci -> !ci.isAbstract() && JandexUtil.resolveTypeParameters(ci.name(), superclass, index)
+                        .stream().anyMatch(t -> t.name().equals(expectedTypeArgument)))
+                .min(Comparator.comparing(ClassInfo::name))
                 .orElse(null);
     }
 
     ClassInfo getImplementorOfWithTypeArgument(DotName implementedInterface, DotName expectedTypeArgument) {
-        return index.getKnownDirectImplementors(implementedInterface)
+        return index.getAllKnownImplementors(implementedInterface)
                 .stream()
-                .filter(ci -> ci.interfaceTypes().stream()
-                        .anyMatch(it -> it.name().equals(implementedInterface)
-                                && it.kind() == Type.Kind.PARAMETERIZED_TYPE
-                                && it.asParameterizedType().arguments().size() == 1
-                                && it.asParameterizedType().arguments().get(0).name().equals(expectedTypeArgument)))
-                .findAny()
+                .filter(ci -> !ci.isAbstract() && JandexUtil.resolveTypeParameters(ci.name(), implementedInterface, index)
+                        .stream().anyMatch(t -> t.name().equals(expectedTypeArgument)))
+                .min(Comparator.comparing(ClassInfo::name))
                 .orElse(null);
     }
 

diff --git a/...t/java/io/quarkus/smallrye/reactivemessaging/kafka/deployment/DefaultSerdeConfigTest.java b/...t/java/io/quarkus/smallrye/reactivemessaging/kafka/deployment/DefaultSerdeConfigTest.java
@@ -86,6 +86,12 @@ private static void doTest(Config customConfig, Tuple[] expectations,
         List<Class<?>> classes = new ArrayList<>(Arrays.asList(classesToIndex));
         classes.add(Incoming.class);
         classes.add(Outgoing.class);
+        classes.add(Serializer.class);
+        classes.add(Deserializer.class);
+        classes.add(io.quarkus.kafka.client.serialization.ObjectMapperDeserializer.class);
+        classes.add(io.quarkus.kafka.client.serialization.ObjectMapperSerializer.class);
+        classes.add(io.quarkus.kafka.client.serialization.JsonbSerializer.class);
+        classes.add(io.quarkus.kafka.client.serialization.JsonbDeserializer.class);
         DefaultSerdeDiscoveryState discovery = new DefaultSerdeDiscoveryState(index(classes)) {
             @Override
             Config getConfig() {
@@ -3001,5 +3007,46 @@ Multi<GenericPayload<ProducerRecord<String, Long>>> method4() {
         }
     }
 
+    @Test
+    void inheritingSerdeClass() {
+        Tuple[] expectations = {
+                tuple("mp.messaging.outgoing.channel1.value.serializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$MyChildSerializer"),
+                tuple("mp.messaging.incoming.channel2.value.deserializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$MyChildDeserializer"),
+        };
+        doTest(expectations, ChannelChildSerializer.class, MyChildSerializer.class, ParentSerializer.class, MyChildDeserializer.class, ParentDeserializer.class);
+    }
+
+    private static class MyChildSerializer extends ParentSerializer<JsonbDto> {
+
+    }
+
+    private static abstract class ParentSerializer<T> implements Serializer<T> {
+
+        @Override
+        public byte[] serialize(String topic, T data) {
+            return new byte[0];
+        }
+    }
+
+    private static class MyChildDeserializer extends ParentDeserializer {
+
+    }
+
+    private static abstract class ParentDeserializer implements Deserializer<JsonbDto> {
+
+        @Override
+        public JsonbDto deserialize(String topic, byte[] data) {
+            return null;
+        }
+    }
+
+    private static class ChannelChildSerializer {
+        @Channel("channel1")
+        Emitter<JsonbDto> emitter1;
+
+        @Channel("channel2")
+        Multi<JsonbDto> channel2;
+    }
+
 
 }
diff --git a/integration-tests/reactive-messaging-kafka/pom.xml b/integration-tests/reactive-messaging-kafka/pom.xml
@@ -40,6 +40,10 @@
             <groupId>io.quarkus</groupId>
             <artifactId>quarkus-resteasy-jackson</artifactId>
         </dependency>
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-jsonb</artifactId>
+        </dependency>
 
         <!-- Health -->
         <dependency>
@@ -140,6 +144,19 @@
                 </exclusion>
             </exclusions>
         </dependency>
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-jsonb-deployment</artifactId>
+            <version>${project.version}</version>
+            <type>pom</type>
+            <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>*</groupId>
+                    <artifactId>*</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
         <dependency>
             <groupId>io.quarkus</groupId>
             <artifactId>quarkus-smallrye-health-deployment</artifactId>