[MGDSTRM-9160] Process authentication and admin client construction as blocking

kafka-admin/pom.xml

@@ -223,6 +223,10 @@
             <groupId>io.quarkus</groupId>
             <artifactId>quarkus-hibernate-validator</artifactId>
         </dependency>
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-kafka-client</artifactId>
+        </dependency>
         <dependency>
             <groupId>io.smallrye.common</groupId>
             <artifactId>smallrye-common-annotation</artifactId>

kafka-admin/src/main/java/org/bf2/admin/kafka/admin/handlers/AdminClientFactory.java

@@ -1,6 +1,5 @@
 package org.bf2.admin.kafka.admin.handlers;
 
-import io.vertx.core.Vertx;
 import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
@@ -25,6 +24,8 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
 
 @RequestScoped
 public class AdminClientFactory {
@@ -39,9 +40,6 @@ public class AdminClientFactory {
     @Inject
     Logger log;
 
-    @Inject
-    Vertx vertx;
-
     @Inject
     KafkaAdminConfigRetriever config;
 
@@ -60,7 +58,7 @@ public class AdminClientFactory {
      * map will be placed in the context under the key identified by the
      * {@link #ADMIN_CLIENT_CONFIG} constant.
      */
-    public AdminClient createAdminClient() {
+    public CompletionStage<AdminClient> createAdminClient() {
         Map<String, Object> acConfig = config.getAcConfig();
 
         if (config.isOauthEnabled()) {
@@ -83,7 +81,7 @@ public AdminClient createAdminClient() {
             log.debug("OAuth is disabled - no attempt to set access token in Admin Client config");
         }
 
-        return AdminClient.create(acConfig);
+        return CompletableFuture.supplyAsync(() -> AdminClient.create(acConfig));
     }
 
     Optional<String> extractCredentials(Optional<String> authorizationHeader) {

kafka-admin/src/main/java/org/bf2/admin/kafka/admin/handlers/OperationsHandler.java

@@ -1,5 +1,6 @@
 package org.bf2.admin.kafka.admin.handlers;
 
+import io.quarkus.runtime.annotations.RegisterForReflection;
 import org.bf2.admin.kafka.admin.Operations;
 import org.bf2.admin.kafka.admin.model.Types;
 import org.eclipse.microprofile.openapi.annotations.Operation;
@@ -29,6 +30,7 @@
 import java.util.Optional;
 import java.util.concurrent.CompletionStage;
 
+@RegisterForReflection
 public interface OperationsHandler {
 
     @POST

kafka-admin/src/main/java/org/bf2/admin/kafka/admin/handlers/RestOperations.java

@@ -128,9 +128,9 @@ public CompletionStage<Response> listTopics(String filter, Types.DeprecatedPageR
                .thenApply(topicList -> Response.ok().entity(topicList).build());
     }
 
-    @Blocking
     @Counted("consume_records_requests")
     @Timed("consume_records_request_time")
+    @Blocking
     public Response consumeRecords(String topicName,
                                    RecordFilterParams params) {
 
@@ -292,16 +292,16 @@ private Pattern filterPattern(String filter) {
     }
 
     <R> CompletionStage<R> withAdminClient(Function<AdminClient, CompletionStage<R>> function) {
-        final AdminClient client = clientFactory.createAdminClient();
-
-        return threadContext.withContextCapture(function.apply(client))
-                .whenComplete((result, error) -> {
-                    try {
-                        client.close();
-                    } catch (Exception e) {
-                        log.warnf("Exception closing Kafka AdminClient", e);
-                    }
-                });
+        return threadContext.withContextCapture(clientFactory.createAdminClient())
+            .thenCompose(client -> function.apply(client).whenComplete((result, error) -> close(client)));
+    }
+
+    void close(AdminClient client) {
+        try {
+            client.close();
+        } catch (Exception e) {
+            log.warnf("Exception closing Kafka AdminClient", e);
+        }
     }
 
     CompletionStage<Response> badRequest(String message) {

kafka-admin/src/main/resources/application.properties

@@ -21,14 +21,13 @@ quarkus.http.ssl.certificate.key-file=${kafka.admin.tls.key:}
 
 # See https://quarkus.io/guides/kafka-dev-services
 # Enable when using quarkus-kafka-client
-#quarkus.kafka.devservices.enabled=false
+quarkus.kafka.devservices.enabled=false
 
-# Remove when quarkus-kafka-client supports Kafka client 3.0
-quarkus.index-dependency.kafka-clients.group-id=org.apache.kafka
-quarkus.index-dependency.kafka-clients.artifact-id=kafka-clients
+quarkus.vertx.max-event-loop-execute-time=4000
 
 # The following properties will be used when adding JWT RBAC provided by quarkus-smallrye-jwt
 quarkus.smallrye-jwt.enabled=true
+quarkus.smallrye-jwt.blocking-authentication=true
 mp.jwt.verify.publickey.location=${kafka.admin.oauth.jwks.endpoint.uri: }
 mp.jwt.verify.issuer=${kafka.admin.oauth.valid.issuer.uri: }
 smallrye.jwt.client.tls.certificate=${kafka.admin.oauth.trusted.cert:}
@@ -44,7 +43,7 @@ quarkus.swagger-ui.theme=monokai
 
 quarkus.log.category."org.apache.kafka".level=WARN
 
-kafka.admin.oauth.enabled=${quarkus.smallrye-jwt.enabled}
+kafka.admin.oauth.enabled=${quarkus.smallrye-jwt.enabled:true}
 # Default limit to the number of partitions that new topics may have configured.
 kafka.admin.num.partitions.max=100
 # Default resource/operations mapping

pom.xml

@@ -28,7 +28,6 @@
         <!-- Dependencies -->
         <quarkus.version>2.7.6.Final</quarkus.version>
         <jandex-maven-plugin.version>1.2.1</jandex-maven-plugin.version>
-        <kafka.version>3.0.0</kafka.version>
         <strimzi-oauth.version>0.8.1</strimzi-oauth.version>
         <smallrye.jwt.version>3.5.3</smallrye.jwt.version>
         <hamcrest.version>2.1</hamcrest.version>
@@ -70,11 +69,6 @@
                 <artifactId>kafka-admin</artifactId>
                 <version>${project.version}</version>
             </dependency>
-            <dependency>
-                <groupId>org.apache.kafka</groupId>
-                <artifactId>kafka-clients</artifactId>
-                <version>${kafka.version}</version>
-            </dependency>
             <dependency>
                 <groupId>io.strimzi</groupId>
                 <artifactId>kafka-oauth-client</artifactId>