Merge pull request #30016 from zakkak/2022-12-21-remove-registration-of-deprecated-kafka-classes

Make registrations of removed (in version 3) kafka classes optional

Author: cescoffier

extensions/kafka-client/deployment/src/main/java/io/quarkus/kafka/client/deployment/KafkaProcessor.java

@@ -228,8 +228,10 @@ public void build(
         collectImplementors(toRegister, indexBuildItem, Serializer.class);
         collectImplementors(toRegister, indexBuildItem, Deserializer.class);
         collectImplementors(toRegister, indexBuildItem, Partitioner.class);
-        // PartitionAssignor is now deprecated, replaced by ConsumerPartitionAssignor
-        collectImplementors(toRegister, indexBuildItem, PARTITION_ASSIGNER);
+        if (QuarkusClassLoader.isClassPresentAtRuntime(PARTITION_ASSIGNER.toString())) {
+            // PartitionAssignor is now deprecated, replaced by ConsumerPartitionAssignor
+            collectImplementors(toRegister, indexBuildItem, PARTITION_ASSIGNER);
+        }
         collectImplementors(toRegister, indexBuildItem, ConsumerPartitionAssignor.class);
         collectImplementors(toRegister, indexBuildItem, ConsumerInterceptor.class);
         collectImplementors(toRegister, indexBuildItem, ProducerInterceptor.class);

extensions/kafka-streams/deployment/src/main/java/io/quarkus/kafka/streams/deployment/KafkaStreamsProcessor.java

@@ -19,6 +19,7 @@
 
 import io.quarkus.arc.deployment.AdditionalBeanBuildItem;
 import io.quarkus.arc.deployment.SyntheticBeanBuildItem;
+import io.quarkus.bootstrap.classloading.QuarkusClassLoader;
 import io.quarkus.deployment.Feature;
 import io.quarkus.deployment.annotations.BuildProducer;
 import io.quarkus.deployment.annotations.BuildStep;
@@ -39,6 +40,8 @@
 
 class KafkaStreamsProcessor {
 
+    public static final String DEFAULT_PARTITION_GROUPER = "org.apache.kafka.streams.processor.DefaultPartitionGrouper";
+
     @BuildStep
     void build(BuildProducer<FeatureBuildItem> feature,
             BuildProducer<ReflectiveClassBuildItem> reflectiveClasses,
@@ -64,9 +67,11 @@ private void registerClassesThatAreLoadedThroughReflection(BuildProducer<Reflect
 
     private void registerCompulsoryClasses(BuildProducer<ReflectiveClassBuildItem> reflectiveClasses) {
         reflectiveClasses.produce(new ReflectiveClassBuildItem(true, false, false, StreamsPartitionAssignor.class));
-        // Class DefaultPartitionGrouper deprecated in Kafka 2.8.x and removed in 3.0.0
-        reflectiveClasses.produce(
-                new ReflectiveClassBuildItem(true, false, false, "org.apache.kafka.streams.processor.DefaultPartitionGrouper"));
+        if (QuarkusClassLoader.isClassPresentAtRuntime(DEFAULT_PARTITION_GROUPER)) {
+            // Class DefaultPartitionGrouper deprecated in Kafka 2.8.x and removed in 3.0.0
+            reflectiveClasses.produce(
+                    new ReflectiveClassBuildItem(true, false, false, DEFAULT_PARTITION_GROUPER));
+        }
         reflectiveClasses.produce(new ReflectiveClassBuildItem(true, false, false, DefaultProductionExceptionHandler.class));
         reflectiveClasses.produce(new ReflectiveClassBuildItem(true, false, false, FailOnInvalidTimestamp.class));
         reflectiveClasses.produce(new ReflectiveClassBuildItem(true, true, true,