Merge the integration test cases for includeFencedBrokers into one so…

… that we don't have to wait for the broker to get fenced twice and use the TestUtils.retry() for it

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

@@ -617,31 +617,19 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
 
   @ParameterizedTest
   @ValueSource(strings = Array("kraft"))
-  def testListNodesNotIncludingFencedBrokers(quorum: String): Unit = {
+  def testListNodesWithIncludeFencedBrokers(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
+    // It takes a few seconds for a broker to get fenced after being killed
+    // So we retry until only 2 of 3 brokers returned in the result or the max wait is reached
+    TestUtils.retry(20000) {
+      assertTrue(client.describeCluster().nodes().get().asScala.size.equals(brokers.size - 1))
+    }
 
+    // List nodes again but this time include the fenced broker
     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)