Repository: quarkusio/quarkus
Issue: Kafka Streams extension fails to compile with Kafka clients 3.0.0

### Describe the bug

Quarkus application BOM doesn't have the Kafka client version 3.0.0 yet. 
When forced to be used with 3.0.0 KafkaStreams extension doesn't compile because of the removal of deprecated classes such as `PartitionAssignor` and `DefaultPartitionGrouper`

### Expected behavior

Kafka extensions should continue to work with Kafka clients version 3.0.0

### Actual behavior

_No response_

### How to Reproduce?

1. In kafka-streams-quickstart/producer, force dependency org.apache.kafka:kafka-clients:3.0.0
2. Compilation fails

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

_No response_

### Output of `java -version`

_No response_

### GraalVM version (if different from Java)

_No response_

### Quarkus version or git rev

_No response_

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

_No response_

### Additional information

_No response_

Changed files:
extensions/kafka-streams/deployment/src/main/java/io/quarkus/kafka/streams/deployment/KafkaStreamsProcessor.java
extensions/kafka-client/deployment/src/main/java/io/quarkus/kafka/client/deployment/KafkaProcessor.java

diff --git a/extensions/kafka-client/deployment/src/main/java/io/quarkus/kafka/client/deployment/KafkaProcessor.java b/extensions/kafka-client/deployment/src/main/java/io/quarkus/kafka/client/deployment/KafkaProcessor.java
index 6ea3fc40bb4..98915dad001 100644
--- a/extensions/kafka-client/deployment/src/main/java/io/quarkus/kafka/client/deployment/KafkaProcessor.java
+++ b/extensions/kafka-client/deployment/src/main/java/io/quarkus/kafka/client/deployment/KafkaProcessor.java
@@ -15,7 +15,6 @@
 import org.apache.kafka.clients.consumer.RangeAssignor;
 import org.apache.kafka.clients.consumer.RoundRobinAssignor;
 import org.apache.kafka.clients.consumer.StickyAssignor;
-import org.apache.kafka.clients.consumer.internals.PartitionAssignor;
 import org.apache.kafka.clients.producer.Partitioner;
 import org.apache.kafka.clients.producer.ProducerInterceptor;
 import org.apache.kafka.clients.producer.internals.DefaultPartitioner;
@@ -123,6 +122,9 @@ public class KafkaProcessor {
             "org.apache.kafka.common.security.oauthbearer.internals.OAuthBearerSaslClientProvider"
     }).collect(Collectors.toSet());
 
+    static final DotName PARTITION_ASSIGNER = DotName
+            .createSimple("org.apache.kafka.clients.consumer.internals.PartitionAssignor");
+
     @BuildStep
     FeatureBuildItem feature() {
         return new FeatureBuildItem(Feature.KAFKA_CLIENT);
@@ -188,7 +190,7 @@ public void build(
         collectImplementors(toRegister, indexBuildItem, Deserializer.class);
         collectImplementors(toRegister, indexBuildItem, Partitioner.class);
         // PartitionAssignor is now deprecated, replaced by ConsumerPartitionAssignor
-        collectImplementors(toRegister, indexBuildItem, PartitionAssignor.class);
+        collectImplementors(toRegister, indexBuildItem, PARTITION_ASSIGNER);
         collectImplementors(toRegister, indexBuildItem, ConsumerPartitionAssignor.class);
         collectImplementors(toRegister, indexBuildItem, ConsumerInterceptor.class);
         collectImplementors(toRegister, indexBuildItem, ProducerInterceptor.class);
@@ -467,6 +469,10 @@ private static void collectImplementors(Set<DotName> set, CombinedIndexBuildItem
         collectClassNames(set, indexBuildItem.getIndex().getAllKnownImplementors(DotName.createSimple(cls.getName())));
     }
 
+    private static void collectImplementors(Set<DotName> set, CombinedIndexBuildItem indexBuildItem, DotName className) {
+        collectClassNames(set, indexBuildItem.getIndex().getAllKnownImplementors(className));
+    }
+
     private static void collectSubclasses(Set<DotName> set, CombinedIndexBuildItem indexBuildItem, Class<?> cls) {
         collectClassNames(set, indexBuildItem.getIndex().getAllKnownSubclasses(DotName.createSimple(cls.getName())));
     }
diff --git a/extensions/kafka-streams/deployment/src/main/java/io/quarkus/kafka/streams/deployment/KafkaStreamsProcessor.java b/extensions/kafka-streams/deployment/src/main/java/io/quarkus/kafka/streams/deployment/KafkaStreamsProcessor.java
index 93774a61fbe..00b6d746a29 100644
--- a/extensions/kafka-streams/deployment/src/main/java/io/quarkus/kafka/streams/deployment/KafkaStreamsProcessor.java
+++ b/extensions/kafka-streams/deployment/src/main/java/io/quarkus/kafka/streams/deployment/KafkaStreamsProcessor.java
@@ -11,7 +11,6 @@
 import org.apache.kafka.streams.StreamsConfig;
 import org.apache.kafka.streams.errors.DefaultProductionExceptionHandler;
 import org.apache.kafka.streams.errors.LogAndFailExceptionHandler;
-import org.apache.kafka.streams.processor.DefaultPartitionGrouper;
 import org.apache.kafka.streams.processor.FailOnInvalidTimestamp;
 import org.apache.kafka.streams.processor.internals.StreamsPartitionAssignor;
 import org.rocksdb.RocksDBException;
@@ -65,7 +64,9 @@ private void registerClassesThatAreLoadedThroughReflection(BuildProducer<Reflect
 
     private void registerCompulsoryClasses(BuildProducer<ReflectiveClassBuildItem> reflectiveClasses) {
         reflectiveClasses.produce(new ReflectiveClassBuildItem(true, false, false, StreamsPartitionAssignor.class));
-        reflectiveClasses.produce(new ReflectiveClassBuildItem(true, false, false, DefaultPartitionGrouper.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"));
         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,