Added check to prevent NPE logs during consumer.close()

Author: ShivsundarR

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java

@@ -895,12 +895,14 @@ private void close(final Duration timeout, final boolean swallowException) {
         closeTimer.update();
 
         // Prepare shutting down the network thread
-        swallow(log, Level.ERROR, "Failed to release assignment before closing consumer",
-                () -> sendAcknowledgementsAndLeaveGroup(closeTimer, firstException), firstException);
-        swallow(log, Level.ERROR, "Failed to stop finding coordinator",
-                this::stopFindCoordinatorOnClose, firstException);
-        swallow(log, Level.ERROR, "Failed invoking acknowledgement commit callback",
-                () -> handleCompletedAcknowledgements(true), firstException);
+        if (applicationEventHandler != null && backgroundEventReaper != null && backgroundEventQueue != null) {
+            swallow(log, Level.ERROR, "Failed to release assignment before closing consumer",
+                    () -> sendAcknowledgementsAndLeaveGroup(closeTimer, firstException), firstException);
+            swallow(log, Level.ERROR, "Failed to stop finding coordinator",
+                    this::stopFindCoordinatorOnClose, firstException);
+            swallow(log, Level.ERROR, "Failed invoking acknowledgement commit callback",
+                    () -> handleCompletedAcknowledgements(true), firstException);
+        }
         if (applicationEventHandler != null)
             closeQuietly(() -> applicationEventHandler.close(Duration.ofMillis(closeTimer.remainingMs())), "Failed shutting down network thread", firstException);
         closeTimer.update();

clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImplTest.java

@@ -43,6 +43,7 @@
 import org.apache.kafka.common.metrics.Metrics;
 import org.apache.kafka.common.protocol.Errors;
 import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.utils.LogCaptureAppender;
 import org.apache.kafka.common.utils.LogContext;
 import org.apache.kafka.common.utils.MockTime;
 import org.apache.kafka.common.utils.Time;
@@ -77,6 +78,7 @@
 import static java.util.Collections.singletonList;
 import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertInstanceOf;
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -207,11 +209,19 @@ public void testFailConstructor() {
         props.put(ConsumerConfig.GROUP_ID_CONFIG, "group-id");
         props.put(ConsumerConfig.METRIC_REPORTER_CLASSES_CONFIG, "an.invalid.class");
         final ConsumerConfig config = new ConsumerConfig(props);
+
+        LogCaptureAppender appender = LogCaptureAppender.createAndRegister();
         KafkaException ce = assertThrows(
                 KafkaException.class,
                 () -> newConsumer(config));
         assertTrue(ce.getMessage().contains("Failed to construct Kafka share consumer"), "Unexpected exception message: " + ce.getMessage());
         assertTrue(ce.getCause().getMessage().contains("Class an.invalid.class cannot be found"), "Unexpected cause: " + ce.getCause());
+
+        boolean npeLogged = appender.getEvents().stream()
+                .flatMap(event -> event.getThrowableInfo().stream())
+                .anyMatch(str -> str.contains("NullPointerException"));
+
+        assertFalse(npeLogged, "Unexpected NullPointerException during consumer construction");
     }
 
     @Test