MINOR: Add interface for aliveBroker and isShutDwon for Brokers. (#16323)

Reviewers: Chia-Ping Tsai <chia7712@gmail.com>

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

@@ -733,6 +733,10 @@ class BrokerServer(
     }
   }
 
+  override def isShutdown(): Boolean = {
+    status == SHUTDOWN || status == SHUTTING_DOWN
+  }
+
   override def awaitShutdown(): Unit = {
     lock.lock()
     try {

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

@@ -97,6 +97,7 @@ trait KafkaBroker extends Logging {
   def awaitShutdown(): Unit
   def shutdown(): Unit = shutdown(Duration.ofMinutes(5))
   def shutdown(timeout: Duration): Unit
+  def isShutdown(): Boolean
   def brokerTopicStats: BrokerTopicStats
   def credentialProvider: CredentialProvider
   def clientToControllerChannelManager: NodeToControllerChannelManager

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

@@ -1085,6 +1085,13 @@ class KafkaServer(
     }
   }
 
+  override def isShutdown(): Boolean = {
+    BrokerState.fromValue(brokerState.value()) match {
+      case BrokerState.SHUTTING_DOWN | BrokerState.NOT_RUNNING => true
+      case _ => false
+    }
+  }
+
   /**
    * After calling shutdown(), use this API to wait until the shutdown is complete
    */

core/src/test/java/kafka/test/ClusterInstance.java

@@ -52,6 +52,11 @@ public interface ClusterInstance {
 
     Map<Integer, KafkaBroker> brokers();
 
+    default Map<Integer, KafkaBroker> aliveBrokers() {
+        return brokers().entrySet().stream().filter(entry -> !entry.getValue().isShutdown())
+                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+    }
+
     Map<Integer, ControllerServer> controllers();
 
     /**

core/src/test/java/kafka/test/ClusterTestExtensionsTest.java

@@ -234,4 +234,19 @@ public class ClusterTestExtensionsTest {
         Assertions.assertTrue(clusterInstance.supportedGroupProtocols().contains(CLASSIC));
         Assertions.assertEquals(1, clusterInstance.supportedGroupProtocols().size());
     }
+
+    @ClusterTest(types = {Type.ZK, Type.CO_KRAFT, Type.KRAFT}, brokers = 4)
+    public void testClusterAliveBrokers(ClusterInstance clusterInstance) throws Exception {
+        clusterInstance.waitForReadyBrokers();
+
+        // Remove broker id 0
+        clusterInstance.shutdownBroker(0);
+        Assertions.assertFalse(clusterInstance.aliveBrokers().containsKey(0));
+        Assertions.assertTrue(clusterInstance.brokers().containsKey(0));
+
+        // add broker id 0 back
+        clusterInstance.startBroker(0);
+        Assertions.assertTrue(clusterInstance.aliveBrokers().containsKey(0));
+        Assertions.assertTrue(clusterInstance.brokers().containsKey(0));
+    }
 }