[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,41 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+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

@@ -17,21 +17,20 @@
 
 package kafka.server
 
-import java.util.{Optional, Properties}
+import java.util.Properties
 
+import integration.kafka.tools.MaintenanceBrokerTestUtils
 import kafka.server.KafkaConfig.fromProps
-import kafka.utils.CoreUtils._
 import kafka.utils.TestUtils
 import kafka.utils.TestUtils._
 import kafka.zk.ZooKeeperTestHarness
 import org.apache.kafka.clients.admin._
 import org.apache.kafka.common.network.ListenerName
 import org.apache.kafka.common.security.auth.SecurityProtocol
-
-import scala.collection.JavaConverters._
 import org.junit.Assert._
 import org.junit.{After, Test}
 
+import scala.collection.JavaConverters._
 import scala.collection.Map
 
 /**
@@ -201,15 +200,7 @@ 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)
   }
 
 }