Repository: quarkusio/quarkus
Issue: Pool AdminClient in KafkaStreams health check

**Describe the bug**

Lots of INFO logs are generated by default due to KafkaStreams health check (500MB of logs in a few hours with 20 pods & 10s check frequency):

```
INFO  <[org.apa.kaf.cli.adm.AdminClientConfig]> AdminClientConfig values: 
	client.dns.lookup = default
	client.id = 
	connections.max.idle.ms = 300000
	(...)
INFO  <[org.apa.kaf.com.uti.AppInfoParser]> version: 2.4.1
```

The logging is kind of hardcoded into Kafka when using [Admin#create](https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/admin/Admin.java#L63), but a workaround is to set:

```
quarkus.log.category."org.apache.kafka.clients.admin.AdminClientConfig".level=WARN
quarkus.log.category."org.apache.kafka.common.utils.AppInfoParser".level=WARN
```

Though creating many admin clients also raises the question of connection usage on the broker.
And we're seeing some AdminClient#close actually taking a long time (not investigated):

```
io.vertx.core.impl.BlockedThread - WARNING <[io.ver.cor.imp.BlockedThreadChecker] TID#17> [PFX: ] Thread Thread[vert.x-worker-thread-5,5,main]=Thread[vert.x-worker-thread-5,5,main] has been blocked for 69273 ms, time limit is 60000 ms: io.vertx.core.VertxException: Thread blocked
	at java.base@11.0.5/java.lang.Object.wait(Native Method)
	at java.base@11.0.5/java.lang.Thread.join(Thread.java:1305)
	at java.base@11.0.5/java.lang.Thread.join(Thread.java:1379)
	at app//org.apache.kafka.clients.admin.KafkaAdminClient.close(KafkaAdminClient.java:541)
	at app//org.apache.kafka.clients.admin.Admin.close(Admin.java:96)
	at app//org.apache.kafka.clients.admin.Admin.close(Admin.java:79)
	at app//io.quarkus.kafka.streams.runtime.KafkaStreamsTopologyManager.getMissingTopics(KafkaStreamsTopologyManager.java:175)
	at app//io.quarkus.kafka.streams.runtime.KafkaStreamsTopologyManager_ClientProxy.getMissingTopics(KafkaStreamsTopologyManager_ClientProxy.zig:339)
	at app//io.quarkus.kafka.streams.runtime.health.KafkaStreamsTopicsHealthCheck.call(KafkaStreamsTopicsHealthCheck.java:50)
	at app//io.quarkus.kafka.streams.runtime.health.KafkaStreamsTopicsHealthCheck_ClientProxy.call(KafkaStreamsTopicsHealthCheck_ClientProxy.zig:117)
	at app//io.smallrye.health.SmallRyeHealthReporter.jsonObject(SmallRyeHealthReporter.java:185)
	at app//io.smallrye.health.SmallRyeHealthReporter.fillCheck(SmallRyeHealthReporter.java:172)
	at app//io.smallrye.health.SmallRyeHealthReporter.processChecks(SmallRyeHealthReporter.java:160)
	at app//io.smallrye.health.SmallRyeHealthReporter.getHealth(SmallRyeHealthReporter.java:138)
	at app//io.smallrye.health.SmallRyeHealthReporter.getReadiness(SmallRyeHealthReporter.java:105)
	at app//io.smallrye.health.SmallRyeHealthReporter_ClientProxy.getReadiness(SmallRyeHealthReporter_ClientProxy.zig:104)
	at app//io.quarkus.smallrye.health.runtime.SmallRyeReadinessHandler.handle(SmallRyeReadinessHandler.java:42)
```

**Expected behavior**

AdminClients should be pooled.
Based on a recent answer to https://stackoverflow.com/questions/57038823/is-adminclient-in-package-kafka-clients-thread-safe , it would appear AdminClient is actually thread-safe (I haven't seen it reflected on the doc yet ?), so we could share the instance.
I'm wondering if other frameworks already took that assumption.

**Actual behavior**

New AdminClient created every 10s. 

**Environment (please complete the following information):**
 - Output of `java -version`:  11.0.5
 - Quarkus version or git rev: 1.4.1.Final


Changed files:
extensions/kafka-streams/runtime/src/main/java/io/quarkus/kafka/streams/runtime/KafkaStreamsTopologyManager.java
extensions/kafka-streams/runtime/src/main/java/io/quarkus/kafka/streams/runtime/KafkaStreamsProducer.java

diff --git a/extensions/kafka-streams/runtime/src/main/java/io/quarkus/kafka/streams/runtime/KafkaStreamsProducer.java b/extensions/kafka-streams/runtime/src/main/java/io/quarkus/kafka/streams/runtime/KafkaStreamsProducer.java
index 8b76e09f2c7..0fc17b028db 100644
--- a/extensions/kafka-streams/runtime/src/main/java/io/quarkus/kafka/streams/runtime/KafkaStreamsProducer.java
+++ b/extensions/kafka-streams/runtime/src/main/java/io/quarkus/kafka/streams/runtime/KafkaStreamsProducer.java
@@ -24,7 +24,7 @@
 import javax.inject.Singleton;
 
 import org.apache.kafka.clients.CommonClientConfigs;
-import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.Admin;
 import org.apache.kafka.clients.admin.AdminClientConfig;
 import org.apache.kafka.clients.admin.ListTopicsResult;
 import org.apache.kafka.common.config.SaslConfigs;
@@ -55,6 +55,7 @@ public class KafkaStreamsProducer {
     private final ExecutorService executorService;
     private final KafkaStreams kafkaStreams;
     private final KafkaStreamsTopologyManager kafkaStreamsTopologyManager;
+    private final Admin kafkaAdminClient;
 
     @Inject
     public KafkaStreamsProducer(KafkaStreamsSupport kafkaStreamsSupport, KafkaStreamsRuntimeConfig runtimeConfig,
@@ -66,6 +67,7 @@ public KafkaStreamsProducer(KafkaStreamsSupport kafkaStreamsSupport, KafkaStream
             this.executorService = null;
             this.kafkaStreams = null;
             this.kafkaStreamsTopologyManager = null;
+            this.kafkaAdminClient = null;
             return;
         }
 
@@ -73,13 +75,13 @@ public KafkaStreamsProducer(KafkaStreamsSupport kafkaStreamsSupport, KafkaStream
 
         String bootstrapServersConfig = asString(runtimeConfig.bootstrapServers);
         Properties kafkaStreamsProperties = getStreamsProperties(buildTimeProperties, bootstrapServersConfig, runtimeConfig);
-        Properties adminClientConfig = getAdminClientConfig(kafkaStreamsProperties);
+        this.kafkaAdminClient = Admin.create(getAdminClientConfig(kafkaStreamsProperties));
 
         this.executorService = Executors.newSingleThreadExecutor();
 
-        this.kafkaStreams = initializeKafkaStreams(kafkaStreamsProperties, runtimeConfig, adminClientConfig, topology.get(),
+        this.kafkaStreams = initializeKafkaStreams(kafkaStreamsProperties, runtimeConfig, kafkaAdminClient, topology.get(),
                 kafkaClientSupplier, stateListener, globalStateRestoreListener, executorService);
-        this.kafkaStreamsTopologyManager = new KafkaStreamsTopologyManager(adminClientConfig);
+        this.kafkaStreamsTopologyManager = new KafkaStreamsTopologyManager(kafkaAdminClient);
     }
 
     @Produces
@@ -107,11 +109,14 @@ void close() {
             LOGGER.debug("Stopping Kafka Streams pipeline");
             kafkaStreams.close();
         }
+        if (kafkaAdminClient != null) {
+            kafkaAdminClient.close();
+        }
     }
 
     private static KafkaStreams initializeKafkaStreams(Properties kafkaStreamsProperties,
-            KafkaStreamsRuntimeConfig runtimeConfig,
-            Properties adminClientConfig, Topology topology, Instance<KafkaClientSupplier> kafkaClientSupplier,
+            KafkaStreamsRuntimeConfig runtimeConfig, Admin adminClient, Topology topology,
+            Instance<KafkaClientSupplier> kafkaClientSupplier,
             Instance<StateListener> stateListener, Instance<StateRestoreListener> globalStateRestoreListener,
             ExecutorService executorService) {
         KafkaStreams kafkaStreams;
@@ -133,7 +138,7 @@ private static KafkaStreams initializeKafkaStreams(Properties kafkaStreamsProper
             @Override
             public void run() {
                 try {
-                    waitForTopicsToBeCreated(adminClientConfig, runtimeConfig.getTrimmedTopics());
+                    waitForTopicsToBeCreated(adminClient, runtimeConfig.getTrimmedTopics());
                 } catch (InterruptedException e) {
                     Thread.currentThread().interrupt();
                     return;
@@ -249,35 +254,33 @@ private static String asString(List<InetSocketAddress> addresses) {
                 .collect(Collectors.joining(","));
     }
 
-    private static void waitForTopicsToBeCreated(Properties adminClientConfig, Collection<String> topicsToAwait)
+    private static void waitForTopicsToBeCreated(Admin adminClient, Collection<String> topicsToAwait)
             throws InterruptedException {
-        try (AdminClient adminClient = AdminClient.create(adminClientConfig)) {
-            Set<String> lastMissingTopics = null;
-            while (true) {
-                try {
-                    ListTopicsResult topics = adminClient.listTopics();
-                    Set<String> existingTopics = topics.names().get(10, TimeUnit.SECONDS);
+        Set<String> lastMissingTopics = null;
+        while (true) {
+            try {
+                ListTopicsResult topics = adminClient.listTopics();
+                Set<String> existingTopics = topics.names().get(10, TimeUnit.SECONDS);
+
+                if (existingTopics.containsAll(topicsToAwait)) {
+                    LOGGER.debug("All expected topics created: " + topicsToAwait);
+                    return;
+                } else {
+                    Set<String> missingTopics = new HashSet<>(topicsToAwait);
+                    missingTopics.removeAll(existingTopics);
 
-                    if (existingTopics.containsAll(topicsToAwait)) {
-                        LOGGER.debug("All expected topics created: " + topicsToAwait);
-                        return;
+                    // Do not spam warnings - topics may take time to be created by an operator like Strimzi
+                    if (missingTopics.equals(lastMissingTopics)) {
+                        LOGGER.debug("Waiting for topic(s) to be created: " + missingTopics);
                     } else {
-                        Set<String> missingTopics = new HashSet<>(topicsToAwait);
-                        missingTopics.removeAll(existingTopics);
-
-                        // Do not spam warnings - topics may take time to be created by an operator like Strimzi
-                        if (missingTopics.equals(lastMissingTopics)) {
-                            LOGGER.debug("Waiting for topic(s) to be created: " + missingTopics);
-                        } else {
-                            LOGGER.warn("Waiting for topic(s) to be created: " + missingTopics);
-                            lastMissingTopics = missingTopics;
-                        }
+                        LOGGER.warn("Waiting for topic(s) to be created: " + missingTopics);
+                        lastMissingTopics = missingTopics;
                     }
-
-                    Thread.sleep(1_000);
-                } catch (ExecutionException | TimeoutException e) {
-                    LOGGER.error("Failed to get topic names from broker", e);
                 }
+
+                Thread.sleep(1_000);
+            } catch (ExecutionException | TimeoutException e) {
+                LOGGER.error("Failed to get topic names from broker", e);
             }
         }
     }
diff --git a/extensions/kafka-streams/runtime/src/main/java/io/quarkus/kafka/streams/runtime/KafkaStreamsTopologyManager.java b/extensions/kafka-streams/runtime/src/main/java/io/quarkus/kafka/streams/runtime/KafkaStreamsTopologyManager.java
index ac696df956a..c33539397ed 100644
--- a/extensions/kafka-streams/runtime/src/main/java/io/quarkus/kafka/streams/runtime/KafkaStreamsTopologyManager.java
+++ b/extensions/kafka-streams/runtime/src/main/java/io/quarkus/kafka/streams/runtime/KafkaStreamsTopologyManager.java
@@ -3,13 +3,12 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashSet;
-import java.util.Properties;
 import java.util.Set;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
-import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.Admin;
 import org.apache.kafka.clients.admin.ListTopicsResult;
 import org.jboss.logging.Logger;
 
@@ -17,16 +16,16 @@ public class KafkaStreamsTopologyManager {
 
     private static final Logger LOGGER = Logger.getLogger(KafkaStreamsTopologyManager.class.getName());
 
-    private final Properties adminClientConfig;
+    private final Admin adminClient;
 
-    public KafkaStreamsTopologyManager(Properties adminClientConfig) {
-        this.adminClientConfig = adminClientConfig;
+    public KafkaStreamsTopologyManager(Admin adminClient) {
+        this.adminClient = adminClient;
     }
 
     public Set<String> getMissingTopics(Collection<String> topicsToCheck) throws InterruptedException {
         HashSet<String> missing = new HashSet<>(topicsToCheck);
 
-        try (AdminClient adminClient = AdminClient.create(adminClientConfig)) {
+        try {
             ListTopicsResult topics = adminClient.listTopics();
             Set<String> topicNames = topics.names().get(10, TimeUnit.SECONDS);