KAFKA-19905: Fix tight reconnection loop during shutdown

This patch fixes a tight reconnection loop that may happen during shutdown.

1. Node 1 and 2 (brokers) request controlled shutdown
2. Controller grants the shutdown
3. Controller itself shuts down (RaftManager shutdown)
4. Node 1 and 2 continue trying to heartbeat to the now-dead controller
5. They get stuck in this reconnection loop because the NodeToControllerRequestThread is still running and hasn't been shut down properly

Signed-off-by: Federico Valeri <[email protected]>

Author: fvaleri

core/src/main/scala/kafka/server/NodeToControllerChannelManager.scala

@@ -317,10 +317,10 @@ class NodeToControllerRequestThread(
   }
 
   override def doWork(): Unit = {
-    val controllerInformation = controllerNodeProvider.getControllerInfo()
     if (activeControllerAddress().isDefined) {
       super.pollOnce(Long.MaxValue)
-    } else {
+    } else if (isRunning()) {
+      val controllerInformation = controllerNodeProvider.getControllerInfo()
       debug("Controller isn't cached, looking for local metadata changes")
       controllerInformation.node match {
         case Some(controllerNode) =>

core/src/test/scala/kafka/server/NodeToControllerRequestThreadTest.scala

@@ -456,6 +456,40 @@ class NodeToControllerRequestThreadTest {
     assertEquals(0, testRequestThread.queueSize)
   }
 
+  @Test
+  def testNoControllerReconnectAfterShutdown(): Unit = {
+    val time = new MockTime()
+    val config = new KafkaConfig(TestUtils.createBrokerConfig(1))
+
+    val metadata = mock(classOf[Metadata])
+    val mockClient = new MockClient(time, metadata)
+
+    val controllerNodeProvider = mock(classOf[ControllerNodeProvider])
+    val controller = new Node(2, "host", 1234)
+
+    when(controllerNodeProvider.getControllerInfo()).thenReturn(controllerInfo(Some(controller)))
+
+    val testRequestThread = new NodeToControllerRequestThread(
+      mockClient, new ManualMetadataUpdater(),
+      controllerNodeProvider, config, time, "", retryTimeoutMs = Long.MaxValue)
+    testRequestThread.started = true
+
+    // Initiate shutdown
+    testRequestThread.initiateShutdown()
+    assertFalse(testRequestThread.isRunning())
+
+    // Track initial invocation count
+    val initialGetControllerInfoCalls = mockingDetails(controllerNodeProvider).getInvocations.size()
+
+    // Call doWork after shutdown
+    // It should not try to get controller info since isRunning() is false
+    testRequestThread.doWork()
+
+    // Verify that getControllerInfo was not called again
+    val finalGetControllerInfoCalls = mockingDetails(controllerNodeProvider).getInvocations.size()
+    assertEquals(initialGetControllerInfoCalls, finalGetControllerInfoCalls)
+  }
+
   private def pollUntil(
     requestThread: NodeToControllerRequestThread,
     condition: () => Boolean,