From dbca7b69be125040526fabc28d3801fb3493a938 Mon Sep 17 00:00:00 2001
From: Guillermo Calvo <guillermo@guillermo.dev>
Date: Thu, 17 Aug 2023 11:59:32 +0200
Subject: [PATCH 1/2] Inhibit TRACE messages for a while to avoid polluting the
 logs

---
 .../processor/KafkaConsumerProcessor.java     | 19 +++++++++++++++++--
 1 file changed, 17 insertions(+), 2 deletions(-)

diff --git a/kafka/src/main/java/io/micronaut/configuration/kafka/processor/KafkaConsumerProcessor.java b/kafka/src/main/java/io/micronaut/configuration/kafka/processor/KafkaConsumerProcessor.java
index b0419066d..a22c86bf5 100644
--- a/kafka/src/main/java/io/micronaut/configuration/kafka/processor/KafkaConsumerProcessor.java
+++ b/kafka/src/main/java/io/micronaut/configuration/kafka/processor/KafkaConsumerProcessor.java
@@ -102,6 +102,7 @@
 
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
+import java.time.Instant;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
@@ -328,8 +329,22 @@ public void close() {
             consumerState.kafkaConsumer.wakeup();
         }
         for (ConsumerState consumerState : consumers.values()) {
-            while (consumerState.closedState == ConsumerCloseState.POLLING) {
-                LOG.trace("consumer not closed yet");
+            if (consumerState.closedState == ConsumerCloseState.POLLING) {
+                final Instant start = Instant.now();
+                Instant silentTime = start;
+                do {
+                    if (LOG.isTraceEnabled()) {
+                        final Instant now = Instant.now();
+                        if (now.isAfter(silentTime)) {
+                            LOG.trace("Consumer {} is not closed yet (waiting {})", consumerState.clientId, Duration.between(start, now));
+                            // Inhibit TRACE messages for a while to avoid polluting the logs
+                            silentTime = now.plusSeconds(5);
+                        }
+                    }
+                } while (consumerState.closedState == ConsumerCloseState.POLLING);
+            }
+            if (LOG.isDebugEnabled()) {
+                LOG.debug("Consumer {} is closed", consumerState.clientId);
             }
         }
         consumers.clear();

From 57f36b66908f0d811b96c1c80117f90767c129a2 Mon Sep 17 00:00:00 2001
From: Guillermo Calvo <guillermo@guillermo.dev>
Date: Thu, 17 Aug 2023 12:03:20 +0200
Subject: [PATCH 2/2] Set test log level to TRACE

---
 kafka/src/test/resources/logback.xml | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/kafka/src/test/resources/logback.xml b/kafka/src/test/resources/logback.xml
index cb4676ec9..31ce8cd88 100644
--- a/kafka/src/test/resources/logback.xml
+++ b/kafka/src/test/resources/logback.xml
@@ -14,5 +14,5 @@
     <!--<logger name="org.apache.kafka" level="TRACE" />-->
 <!--    <logger name="io.micronaut.context" level="TRACE"/>-->
     <!--<logger name="io.micronaut.configuration.kafka.processor" level="TRACE"/>-->
-<!--    <logger name="io.micronaut.configuration.kafka " level="TRACE" />-->
+    <logger name="io.micronaut.configuration.kafka" level="TRACE" />
 </configuration>