Fix the issue reported in the review comment

Rebase to trunk and resolve conflicts
Refine the tests to make sure that brokers have the expected fenced state
Add tests for listing nodes from controller (this needed to be in SSLAdminIntegrationTest because of how useBoostrapControllers is set up)

clients/src/main/java/org/apache/kafka/common/requests/DescribeClusterResponse.java

@@ -39,7 +39,7 @@ public DescribeClusterResponse(DescribeClusterResponseData data) {
 
     public Map<Integer, Node> nodes() {
         return data.brokers().valuesList().stream()
-            .map(b -> new Node(b.brokerId(), b.host(), b.port(), b.rack()))
+            .map(b -> new Node(b.brokerId(), b.host(), b.port(), b.rack(), b.isFenced()))
             .collect(Collectors.toMap(Node::id, Function.identity()));
     }
 

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

@@ -615,6 +615,42 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
     assertEquals(brokerStrs.mkString(","), nodeStrs.mkString(","))
   }
 
+  @ParameterizedTest
+  @ValueSource(strings = Array("kraft"))
+  def testListNodesNotIncludingFencedBrokers(quorum: String): Unit = {
+    client = createAdminClient
+    val fencedBrokerId = brokers.last.config.brokerId
+    killBroker(fencedBrokerId)
+    Thread.sleep(10000) //sleep is needed to ensure the broker is fenced after the shutdown
+
+    val nodes = client.describeCluster().nodes().get().asScala
+    assertTrue(nodes.size.equals(brokers.size - 1))
+    nodes.foreach(node => {
+      assertFalse(node.isFenced)
+      assertFalse(node.id().equals(fencedBrokerId))
+    })
+  }
+
+  @ParameterizedTest
+  @ValueSource(strings = Array("kraft"))
+  def testListNodesIncludingFencedBrokers(quorum: String): Unit = {
+    client = createAdminClient
+    val fencedBrokerId = brokers.last.config.brokerId
+    killBroker(fencedBrokerId)
+    Thread.sleep(10000) //sleep is needed to ensure the broker is fenced after the shutdown
+
+    val nodes = client.describeCluster(new DescribeClusterOptions().includeFencedBrokers(true)).nodes().get().asScala
+    assertTrue(nodes.size.equals(brokers.size))
+
+    nodes.foreach(node => {
+      if (node.id().equals(fencedBrokerId)) {
+        assertTrue(node.isFenced)
+      } else {
+        assertFalse(node.isFenced)
+      }
+    })
+  }
+
   @ParameterizedTest
   @ValueSource(strings = Array("kraft"))
   def testAdminClientHandlingBadIPWithoutTimeout(quorum: String): Unit = {

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

@@ -160,20 +160,21 @@ class SslAdminIntegrationTest extends SaslSslAdminIntegrationTest {
 
   @ParameterizedTest
   @ValueSource(strings = Array("kraft"))
-  def testDescribeClusterIncludingFencedBrokers(quorum: String): Unit = {
+  def testListNodesFromControllersIncludingFencedBrokers(quorum: String): Unit = {
+    useBoostrapControllers()
     client = createAdminClient
-    val result = client.describeCluster(new DescribeClusterOptions().includeFencedBrokers(true));
-    assertTrue(result.nodes().get().size().equals(3))
+    val result = client.describeCluster(new DescribeClusterOptions().includeFencedBrokers(true))
+    val exception = assertThrows(classOf[Exception], () => { result.nodes().get()})
+    assertTrue(exception.getCause.getCause.getMessage.contains("Cannot request fenced brokers from controller endpoint"))
   }
 
   @ParameterizedTest
   @ValueSource(strings = Array("kraft"))
-  def testDescribeClusterFromControllersIncludingFencedBrokers(quorum: String): Unit = {
+  def testListNodesFromControllers(quorum: String): Unit = {
     useBoostrapControllers()
     client = createAdminClient
-    val result = client.describeCluster(new DescribeClusterOptions().includeFencedBrokers(true));
-    val exception = assertThrows(classOf[Exception], () => { result.nodes().get()})
-    assertTrue(exception.getCause.getCause.getMessage.contains("Cannot request fenced brokers from controller endpoint"))
+    val result = client.describeCluster(new DescribeClusterOptions())
+    assertTrue(result.nodes().get().size().equals(controllerServers.size))
   }
 
   @ParameterizedTest