Bump kafka-clients from 3.9.0 to 4.0.0

Author: ozangunalp

bom/application/pom.xml

@@ -139,7 +139,7 @@
         <jboss-logging.version>3.6.1.Final</jboss-logging.version>
         <mutiny.version>2.9.0</mutiny.version>
         <jctools-core.version>4.0.5</jctools-core.version>
-        <kafka3.version>3.9.0</kafka3.version>
+        <kafka3.version>4.0.0</kafka3.version>
         <lz4.version>1.8.0</lz4.version> <!-- dependency of the kafka-clients that could be overridden by other imported BOMs in the platform -->
         <snappy.version>1.1.10.5</snappy.version>
         <strimzi-test-container.version>0.109.1</strimzi-test-container.version>

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

@@ -20,7 +20,7 @@
 import org.apache.kafka.clients.consumer.StickyAssignor;
 import org.apache.kafka.clients.producer.Partitioner;
 import org.apache.kafka.clients.producer.ProducerInterceptor;
-import org.apache.kafka.clients.producer.internals.DefaultPartitioner;
+import org.apache.kafka.common.metrics.MetricsReporter;
 import org.apache.kafka.common.security.auth.AuthenticateCallbackHandler;
 import org.apache.kafka.common.security.auth.SecurityProtocol;
 import org.apache.kafka.common.security.authenticator.AbstractLogin;
@@ -249,6 +249,7 @@ public void build(
         collectImplementors(toRegister, indexBuildItem, ConsumerPartitionAssignor.class);
         collectImplementors(toRegister, indexBuildItem, ConsumerInterceptor.class);
         collectImplementors(toRegister, indexBuildItem, ProducerInterceptor.class);
+        collectImplementors(toRegister, indexBuildItem, MetricsReporter.class);
 
         reflectiveClass.produce(ReflectiveClassBuildItem.builder(OAuthBearerSaslClient.class,
                 OAuthBearerSaslClient.OAuthBearerSaslClientFactory.class,
@@ -295,7 +296,6 @@ public void build(
 
         // built in partitioner and partition assignors
         reflectiveClass.produce(ReflectiveClassBuildItem.builder(
-                DefaultPartitioner.class,
                 RangeAssignor.class,
                 RoundRobinAssignor.class,
                 StickyAssignor.class)

extensions/kafka-client/runtime/src/main/java/io/quarkus/kafka/client/runtime/graal/Crc32CSubstitutions.java

@@ -57,8 +57,8 @@ public void createTopic(String topic, int partitions) {
     public int partitions(String topic) {
         TopicDescription topicDescription;
         try {
-            Map<String, TopicDescription> partitions = admin.describeTopics(Collections.singletonList(topic)).all()
-                    .get(2000, TimeUnit.MILLISECONDS);
+            Map<String, TopicDescription> partitions = admin.describeTopics(Collections.singletonList(topic))
+                    .allTopicNames().get(2000, TimeUnit.MILLISECONDS);
             topicDescription = partitions.get(topic);
         } catch (InterruptedException | ExecutionException | TimeoutException e) {
             throw new RuntimeException(e);

integration-tests/compose-devservices/src/main/java/io/quarkus/it/compose/devservices/kafka/KafkaAdminManager.java

@@ -43,8 +43,8 @@ void cleanup() {
     public int partitions(String topic) {
         TopicDescription topicDescription;
         try {
-            Map<String, TopicDescription> partitions = admin.describeTopics(Collections.singletonList(topic)).all()
-                    .get(2000, TimeUnit.MILLISECONDS);
+            Map<String, TopicDescription> partitions = admin.describeTopics(Collections.singletonList(topic))
+                    .allTopicNames().get(2000, TimeUnit.MILLISECONDS);
             topicDescription = partitions.get(topic);
         } catch (InterruptedException | ExecutionException | TimeoutException e) {
             throw new RuntimeException(e);

integration-tests/kafka-devservices/src/main/java/io/quarkus/it/kafka/KafkaAdminManager.java

@@ -177,7 +177,6 @@
                         <artifactId>maven-surefire-plugin</artifactId>
                         <configuration>
                             <skip>false</skip>
-                            <argLine>-Djava.security.manager=allow</argLine>
                             <systemPropertyVariables>
                                 <!-- Configure the app to test to resolve "keycloak" hostname to the Docker hostname -->
                                 <keycloak.docker.image>${keycloak.docker.image}</keycloak.docker.image>
@@ -189,7 +188,6 @@
                         <artifactId>maven-failsafe-plugin</artifactId>
                         <configuration>
                             <skip>false</skip>
-                            <argLine>-Djava.security.manager=allow</argLine>
                             <systemPropertyVariables>
                                 <keycloak.docker.image>${keycloak.docker.image}</keycloak.docker.image>
                                 <keycloak.realm.json>${project.basedir}/src/test/resources/keycloak/realms/kafka-authz-realm.json</keycloak.realm.json>