KAFKA-18569: New consumer close may wait on unneeded FindCoordinator

JIRA: KAFKA-18569
Please refer to ticker for further details

Author: frankvicky

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

@@ -59,6 +59,7 @@ public class CoordinatorRequestManager implements RequestManager {
     private final RequestState coordinatorRequestState;
     private long timeMarkedUnknownMs = -1L; // starting logging a warning only after unable to connect for a while
     private long totalDisconnectedMin = 0;
+    private boolean closing = false;
     private Node coordinator;
 
     public CoordinatorRequestManager(
@@ -92,7 +93,7 @@ public CoordinatorRequestManager(
      */
     @Override
     public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
-        if (this.coordinator != null)
+        if (closing || this.coordinator != null)
             return EMPTY;
 
         if (coordinatorRequestState.canSendRequest(currentTimeMs)) {
@@ -103,6 +104,12 @@ public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
         return new NetworkClientDelegate.PollResult(coordinatorRequestState.remainingBackoffMs(currentTimeMs));
     }
 
+    @Override
+    public NetworkClientDelegate.PollResult pollOnClose(long currentTimeMs) {
+        closing = true;
+        return EMPTY;
+    }
+
     NetworkClientDelegate.UnsentRequest makeFindCoordinatorRequest(final long currentTimeMs) {
         coordinatorRequestState.onSendAttempt(currentTimeMs);
         FindCoordinatorRequestData data = new FindCoordinatorRequestData()

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

@@ -254,6 +254,26 @@ public void testNetworkTimeout() {
         assertEquals(1, res2.unsentRequests.size());
     }
 
+    @Test
+    public void testSignalOnClose() {
+        CoordinatorRequestManager coordinatorManager = setupCoordinatorManager(GROUP_ID);
+
+        expectFindCoordinatorRequest(coordinatorManager, Errors.NONE);
+        assertTrue(coordinatorManager.coordinator().isPresent());
+
+        coordinatorManager.markCoordinatorUnknown("coordinator changed", time.milliseconds());
+        assertEquals(Collections.emptyList(), coordinatorManager.poll(time.milliseconds()).unsentRequests);
+
+        coordinatorManager.pollOnClose(time.milliseconds());
+
+        time.sleep(RETRY_BACKOFF_MS - 1);
+        assertEquals(Collections.emptyList(), coordinatorManager.poll(time.milliseconds()).unsentRequests);
+
+        time.sleep(RETRY_BACKOFF_MS);
+        assertEquals(Collections.emptyList(), coordinatorManager.poll(time.milliseconds()).unsentRequests,
+            "Should not generate find coordinator request during close");
+    }
+
     private void expectFindCoordinatorRequest(
         CoordinatorRequestManager  coordinatorManager,
         Errors error

core/src/test/scala/integration/kafka/api/SaslClientsWithInvalidCredentialsTest.scala

@@ -122,7 +122,7 @@ class SaslClientsWithInvalidCredentialsTest extends AbstractSaslTest {
   }
 
   @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
-  @MethodSource(Array("getTestQuorumAndGroupProtocolParametersClassicGroupProtocolOnly"))
+  @MethodSource(Array("getTestQuorumAndGroupProtocolParametersConsumerGroupProtocolOnly"))
   def testConsumerWithAuthenticationFailure(quorum: String, groupProtocol: String): Unit = {
     val consumer = createConsumer()
     consumer.subscribe(List(topic).asJava)