Avoid log pollution while waiting for consumer to close (#821)

* Inhibit TRACE messages for a while to avoid polluting the logs * Set test log level to TRACE
micronaut-projects · Aug 17, 2023 · b0d144b · b0d144b
1 parent 44d7b8f
commit b0d144b
Show file tree

Hide file tree

Showing 2 changed files with 18 additions and 3 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
@@ -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();

diff --git 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>