Repository: quarkusio/quarkus
Issue: Kafka Streams - Missing topic holds app execution but there is no warning

**Describe the bug**
(Describe the problem clearly and concisely.)
If topics are  declared in the application.properties, the admin client waits for them to be created to start running the kafka streams application. There is no message in console or log about this behavior.

**Expected behavior**
(Describe the expected behavior clearly and concisely.)
Should produce a WARN mentioning app can't start processing messages because topic is missing.

**Actual behavior**
(Describe the actual behavior clearly and concisely.)
App freezes without any feeback

**To Reproduce**
Steps to reproduce the behavior:
1. Add non existent topic to properties
2. Start kafka streams app
3. 

**Configuration**
```properties
# Add your application.properties here, if applicable.

```

**Screenshots**
(If applicable, add screenshots to help explain your problem.)

**Environment (please complete the following information):**
 - Output of `uname -a` or `ver`: 
 - Output of `java -version`: 
 - GraalVM version (if different from Java): 
 - Quarkus version or git rev: 

**Additional context**
(Add any other context about the problem here.)


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

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 7c82607b0f2..c083ea657ec 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
@@ -160,18 +160,26 @@ public KafkaStreams getStreams() {
     private void waitForTopicsToBeCreated(Collection<String> topicsToAwait)
             throws InterruptedException {
         try (AdminClient adminClient = AdminClient.create(adminClientConfig)) {
+            Set<String> lastMissingTopics = null;
             while (true) {
                 try {
                     ListTopicsResult topics = adminClient.listTopics();
-                    Set<String> topicNames = topics.names().get(10, TimeUnit.SECONDS);
+                    Set<String> existingTopics = topics.names().get(10, TimeUnit.SECONDS);
 
-                    if (topicNames.containsAll(topicsToAwait)) {
-                        LOGGER.debug("All expected topics created");
+                    if (existingTopics.containsAll(topicsToAwait)) {
+                        LOGGER.debug("All expected topics created: " + topicsToAwait);
                         return;
                     } else {
-                        Set<String> missing = new HashSet<>(topicsToAwait);
-                        missing.removeAll(topicNames);
-                        LOGGER.debug("Waiting for topic(s) to be created: " + missing);
+                        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;
+                        }
                     }
 
                     Thread.sleep(1_000);