Repository: quarkusio/quarkus
Issue: Kafka configuration reported as unknown config

### Describe the bug

I have configured a property for a kafka incoming topic which quarkus report as unknown:
`mp.messaging.incoming.topic-test1-in.pause-if-no-requests=false`

The documentation has this propertties available to use but I get a warning on startup:
`2022-09-13 17:34:03,125 WARN  [org.apa.kaf.cli.con.ConsumerConfig] (main) The configuration 'pause-if-no-requests' was supplied but isn't a known config.`

In DEV mode, sometimes, it also issues a warning for the property:
`mp.messaging.incoming.topic-test1-in.group.id=my-consumer-group`
The warning is issues on the AdminClientConfig class:
`2022-09-13 17:38:29,082 WARN  [org.apa.kaf.cli.adm.AdminClientConfig] (executor-thread-0) The configuration 'group.id' was supplied but isn't a known config`


### Expected behavior

No warning issued if these property is present

### Actual behavior

_No response_

### How to Reproduce?

_No response_

### 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

2.12.1.Final

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

_No response_

### Additional information

_No response_

Changed files:
extensions/kafka-client/runtime/src/main/java/io/quarkus/kafka/client/runtime/ui/KafkaTopicClient.java
extensions/kafka-client/runtime/src/main/java/io/quarkus/kafka/client/runtime/KafkaAdminClient.java
extensions/kafka-client/deployment/src/main/java/io/quarkus/kafka/client/deployment/DevServicesKafkaProcessor.java

diff --git a/extensions/kafka-client/deployment/src/main/java/io/quarkus/kafka/client/deployment/DevServicesKafkaProcessor.java b/extensions/kafka-client/deployment/src/main/java/io/quarkus/kafka/client/deployment/DevServicesKafkaProcessor.java
index 0bbe4d81122..03b704e56fe 100644
--- a/extensions/kafka-client/deployment/src/main/java/io/quarkus/kafka/client/deployment/DevServicesKafkaProcessor.java
+++ b/extensions/kafka-client/deployment/src/main/java/io/quarkus/kafka/client/deployment/DevServicesKafkaProcessor.java
@@ -152,7 +152,7 @@ public void createTopicPartitions(String bootstrapServers, KafkaDevServiceCfg co
             // get current partitions for topics asked to be created
             Set<String> currentTopics = adminClient.listTopics().names()
                     .get(adminClientTimeout, TimeUnit.MILLISECONDS);
-            Map<String, TopicDescription> partitions = adminClient.describeTopics(currentTopics).all()
+            Map<String, TopicDescription> partitions = adminClient.describeTopics(currentTopics).allTopicNames()
                     .get(adminClientTimeout, TimeUnit.MILLISECONDS);
             // find new topics to create
             List<NewTopic> newTopics = topicPartitions.entrySet().stream()
diff --git a/extensions/kafka-client/runtime/src/main/java/io/quarkus/kafka/client/runtime/KafkaAdminClient.java b/extensions/kafka-client/runtime/src/main/java/io/quarkus/kafka/client/runtime/KafkaAdminClient.java
index c9b75dc1d00..d9901b7bb12 100644
--- a/extensions/kafka-client/runtime/src/main/java/io/quarkus/kafka/client/runtime/KafkaAdminClient.java
+++ b/extensions/kafka-client/runtime/src/main/java/io/quarkus/kafka/client/runtime/KafkaAdminClient.java
@@ -30,8 +30,13 @@ public class KafkaAdminClient {
 
     @PostConstruct
     void init() {
-        Map<String, Object> conf = new HashMap<>(config);
+        Map<String, Object> conf = new HashMap<>();
         conf.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, DEFAULT_ADMIN_CLIENT_TIMEOUT);
+        for (Map.Entry<String, Object> entry : config.entrySet()) {
+            if (AdminClientConfig.configNames().contains(entry.getKey())) {
+                conf.put(entry.getKey(), entry.getValue().toString());
+            }
+        }
         client = AdminClient.create(conf);
     }
 
@@ -81,4 +86,11 @@ public Collection<AclBinding> getAclInfo() throws InterruptedException, Executio
         var options = new DescribeAclsOptions().timeoutMs(1_000);
         return client.describeAcls(filter, options).values().get();
     }
+
+    public Map<String, TopicDescription> describeTopics(Collection<String> topicNames)
+            throws InterruptedException, ExecutionException {
+        return client.describeTopics(topicNames)
+                .allTopicNames()
+                .get();
+    }
 }
diff --git a/extensions/kafka-client/runtime/src/main/java/io/quarkus/kafka/client/runtime/ui/KafkaTopicClient.java b/extensions/kafka-client/runtime/src/main/java/io/quarkus/kafka/client/runtime/ui/KafkaTopicClient.java
index 174ef04aa08..8a2340fd475 100644
--- a/extensions/kafka-client/runtime/src/main/java/io/quarkus/kafka/client/runtime/ui/KafkaTopicClient.java
+++ b/extensions/kafka-client/runtime/src/main/java/io/quarkus/kafka/client/runtime/ui/KafkaTopicClient.java
@@ -9,12 +9,9 @@
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
-import javax.annotation.PostConstruct;
 import javax.enterprise.context.ApplicationScoped;
 import javax.inject.Inject;
 
-import org.apache.kafka.clients.admin.AdminClient;
-import org.apache.kafka.clients.admin.AdminClientConfig;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
@@ -27,6 +24,7 @@
 import org.apache.kafka.common.serialization.BytesSerializer;
 import org.apache.kafka.common.utils.Bytes;
 
+import io.quarkus.kafka.client.runtime.KafkaAdminClient;
 import io.quarkus.kafka.client.runtime.ui.model.Order;
 import io.quarkus.kafka.client.runtime.ui.model.converter.KafkaModelConverter;
 import io.quarkus.kafka.client.runtime.ui.model.request.KafkaMessageCreateRequest;
@@ -38,8 +36,8 @@ public class KafkaTopicClient {
     // TODO: make configurable
     private static final int RETRIES = 3;
 
-    //TODO: inject me
-    private AdminClient adminClient;
+    @Inject
+    KafkaAdminClient adminClient;
 
     KafkaModelConverter modelConverter = new KafkaModelConverter();
 
@@ -47,13 +45,6 @@ public class KafkaTopicClient {
     @Identifier("default-kafka-broker")
     Map<String, Object> config;
 
-    @PostConstruct
-    void init() {
-        Map<String, Object> conf = new HashMap<>(config);
-        conf.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, "5000");
-        adminClient = AdminClient.create(conf);
-    }
-
     private Producer<Bytes, Bytes> createProducer() {
         Map<String, Object> config = new HashMap<>(this.config);
 
@@ -260,8 +251,6 @@ public void createMessage(KafkaMessageCreateRequest request) {
 
     public List<Integer> partitions(String topicName) throws ExecutionException, InterruptedException {
         return adminClient.describeTopics(List.of(topicName))
-                .allTopicNames()
-                .get()
                 .values().stream()
                 .reduce((a, b) -> {
                     throw new IllegalStateException(