MINOR: Port changes from KAFKA-18569 for ShareConsumers (apache#19402)

ShareConsumers` may wait on an unneeded `FindCoordinator` during
`close()`(i.e after the acknowledgements are sent).

apache#18590 added the
`StopFindCoordinatorOnClose` event and was used by the regular
consumers. We are using the same event in `ShareConsumers` as well to
prevent sending this event when coordinator is no longer needed.

Reviewers: Andrew Schofield <[email protected]>

Author: ShivsundarR

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

@@ -47,6 +47,7 @@
 import org.apache.kafka.clients.consumer.internals.events.ShareFetchEvent;
 import org.apache.kafka.clients.consumer.internals.events.ShareSubscriptionChangeEvent;
 import org.apache.kafka.clients.consumer.internals.events.ShareUnsubscribeEvent;
+import org.apache.kafka.clients.consumer.internals.events.StopFindCoordinatorOnCloseEvent;
 import org.apache.kafka.clients.consumer.internals.metrics.AsyncConsumerMetrics;
 import org.apache.kafka.clients.consumer.internals.metrics.KafkaShareConsumerMetrics;
 import org.apache.kafka.common.KafkaException;
@@ -887,6 +888,8 @@ private void close(final Duration timeout, final boolean swallowException) {
         // 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",
                 this::handleCompletedAcknowledgements, firstException);
         if (applicationEventHandler != null)
@@ -915,6 +918,11 @@ private void close(final Duration timeout, final boolean swallowException) {
         }
     }
 
+    private void stopFindCoordinatorOnClose() {
+        log.debug("Stop finding coordinator during consumer close");
+        applicationEventHandler.add(new StopFindCoordinatorOnCloseEvent());
+    }
+
     private Timer createTimerForCloseRequests(Duration timeout) {
         // this.time could be null if an exception occurs in constructor prior to setting the this.time field
         final Time time = (this.time == null) ? Time.SYSTEM : this.time;

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

@@ -29,6 +29,7 @@
 import org.apache.kafka.clients.consumer.internals.events.ShareFetchEvent;
 import org.apache.kafka.clients.consumer.internals.events.ShareSubscriptionChangeEvent;
 import org.apache.kafka.clients.consumer.internals.events.ShareUnsubscribeEvent;
+import org.apache.kafka.clients.consumer.internals.events.StopFindCoordinatorOnCloseEvent;
 import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.TopicIdPartition;
 import org.apache.kafka.common.TopicPartition;
@@ -49,6 +50,7 @@
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentMatchers;
+import org.mockito.InOrder;
 import org.mockito.Mockito;
 
 import java.time.Duration;
@@ -78,6 +80,7 @@
 import static org.mockito.ArgumentMatchers.argThat;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.inOrder;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
@@ -351,6 +354,25 @@ public void testCloseWithTopicAuthorizationException() {
         assertDoesNotThrow(() -> consumer.close());
     }
 
+    @Test
+    public void testStopFindCoordinatorOnClose() {
+        SubscriptionState subscriptions = new SubscriptionState(new LogContext(), AutoOffsetResetStrategy.NONE);
+        consumer = newConsumer(subscriptions);
+
+        // Setup the expected successful completion of close events
+        completeShareAcknowledgeOnCloseApplicationEventSuccessfully();
+        completeShareUnsubscribeApplicationEventSuccessfully(subscriptions);
+
+        // Close the consumer
+        consumer.close();
+
+        // Verify events are sent in correct order using InOrder
+        InOrder inOrder = inOrder(applicationEventHandler);
+        inOrder.verify(applicationEventHandler).addAndGet(any(ShareAcknowledgeOnCloseEvent.class));
+        inOrder.verify(applicationEventHandler).add(any(ShareUnsubscribeEvent.class));
+        inOrder.verify(applicationEventHandler).add(any(StopFindCoordinatorOnCloseEvent.class));
+    }
+
     @Test
     public void testVerifyApplicationEventOnShutdown() {
         SubscriptionState subscriptions = new SubscriptionState(new LogContext(), AutoOffsetResetStrategy.NONE);