[LI-HOTFIX] Add test case for describe maintenance broker config

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

@@ -26,9 +26,10 @@ import java.util.{Collections, Optional, Properties}
 import java.{time, util}
 
 import integration.kafka.api.BaseAdminIntegrationTest
+import integration.kafka.tools.MaintenanceBrokerTestUtils
 import kafka.log.LogConfig
 import kafka.security.auth.Group
-import kafka.server.{Defaults, KafkaConfig, KafkaServer}
+import kafka.server.{Defaults, DynamicConfig, KafkaConfig, KafkaServer}
 import kafka.utils.TestUtils._
 import kafka.utils.{Log4jController, TestUtils}
 import kafka.zk.KafkaZkClient
@@ -891,6 +892,23 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
     assertTrue(intercept[ExecutionException](describeResult2.values.get(invalidTopic).get).getCause.isInstanceOf[InvalidTopicException])
   }
 
+  @Test
+  def testDescribeConfigsForMaintenanceBroker(): Unit = {
+    client = AdminClient.create(createConfig)
+    val maintenanceBrokerIds = Seq(0, 2)
+    MaintenanceBrokerTestUtils.setMaintenanceBrokers(adminZkClient, zkClient, servers, maintenanceBrokerIds)
+    servers.map(_.config.brokerId).foreach { brokerId =>
+      val describedMaintenanceBrokerConfigStr =
+        client.describeConfigs(Collections.singleton(new ConfigResource(ConfigResource.Type.BROKER, brokerId.toString)))
+          .all().get(5, TimeUnit.SECONDS)
+          .values().asScala.head.get(DynamicConfig.Broker.MaintenanceBrokerListProp)
+          .value()
+      assertNotNull(describedMaintenanceBrokerConfigStr)
+      val describedMaintenanceBrokerIds = describedMaintenanceBrokerConfigStr.split(",").map(_.toInt).toSet
+      assertEquals(describedMaintenanceBrokerIds, maintenanceBrokerIds.toSet)
+    }
+  }
+
   private def subscribeAndWaitForAssignment(topic: String, consumer: KafkaConsumer[Array[Byte], Array[Byte]]): Unit = {
     consumer.subscribe(Collections.singletonList(topic))
     TestUtils.pollUntilTrue(consumer, () => !consumer.assignment.isEmpty, "Expected non-empty assignment")

core/src/test/scala/integration/kafka/tools/MaintenanceBrokerTestUtils.scala

@@ -0,0 +1,24 @@
+package integration.kafka.tools
+
+import kafka.server.{DynamicConfig, KafkaServer}
+import kafka.utils.CoreUtils.propsWith
+import kafka.utils.TestUtils
+import kafka.zk.{AdminZkClient, KafkaZkClient}
+
+object MaintenanceBrokerTestUtils {
+
+  def setMaintenanceBrokers(adminZkClient: AdminZkClient,
+                            zkClient: KafkaZkClient,
+                            brokers: Seq[KafkaServer],
+                            maintenanceBrokerIds: Seq[Int]): Unit = {
+    val propstring = maintenanceBrokerIds.mkString(",")
+    adminZkClient.changeBrokerConfig(None,
+      propsWith((DynamicConfig.Broker.MaintenanceBrokerListProp, propstring)))
+
+    val controllerId = TestUtils.waitUntilControllerElected(zkClient)
+
+    TestUtils.waitUntilTrue(() => brokers(controllerId).config.getMaintenanceBrokerList == maintenanceBrokerIds,
+      s"wait until broker $propstring is masked as maintenance broker not taking new partitions", 5000)
+  }
+
+}

core/src/test/scala/unit/kafka/server/MaintenanceBrokerTest.scala

@@ -19,6 +19,7 @@ package kafka.server
 
 import java.util.{Optional, Properties}
 
+import integration.kafka.tools.MaintenanceBrokerTestUtils
 import kafka.server.KafkaConfig.fromProps
 import kafka.utils.CoreUtils._
 import kafka.utils.TestUtils
@@ -201,15 +202,6 @@ class MaintenanceBrokerTest extends ZooKeeperTestHarness {
   }
 
   def setMaintenanceBrokers(brokerIds: Seq[Int]): Unit = {
-    var propstring = brokerIds.mkString(",")
-    adminZkClient.changeBrokerConfig(None,
-      propsWith((DynamicConfig.Broker.MaintenanceBrokerListProp, propstring)))
-
-    val controllerId = TestUtils.waitUntilControllerElected(zkClient)
-
-    TestUtils.waitUntilTrue(() => brokers(controllerId).config.getMaintenanceBrokerList == brokerIds,
-      s"wait until broker $propstring is masked as maintenance broker not taking new partitions", 5000)
-
+    MaintenanceBrokerTestUtils.setMaintenanceBrokers(adminZkClient, zkClient, brokers, brokerIds)
   }
-
 }