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

@@ -220,7 +220,7 @@ class NodeToControllerRequestThread(
   initialNetworkClient,
   Math.min(Int.MaxValue, Math.min(config.controllerSocketTimeoutMs, retryTimeoutMs)).toInt,
   time,
-  false
+  true
 ) with Logging {
 
   this.logIdent = logPrefix

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

@@ -456,6 +456,35 @@ class NodeToControllerRequestThreadTest {
     assertEquals(0, testRequestThread.queueSize)
   }
 
+  @Test
+  def testThreadInterruptibleDuringShutdown(): 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)
+
+    // Start the thread
+    testRequestThread.start()
+    TestUtils.waitUntilTrue(() => testRequestThread.isStarted, "Thread did not start", 5000)
+
+    // Shutdown the thread (should interrupt it since isInterruptible=true)
+    testRequestThread.shutdown()
+
+    // Verify shutdown completed successfully
+    assertTrue(testRequestThread.isShutdownComplete())
+    assertFalse(testRequestThread.isRunning())
+  }
+
   private def pollUntil(
     requestThread: NodeToControllerRequestThread,
     condition: () => Boolean,