diff --git a/core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala b/core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala index b96a1024a6c73..3902d63d4de38 100644 --- a/core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala +++ b/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") diff --git a/core/src/test/scala/integration/kafka/tools/MaintenanceBrokerTestUtils.scala b/core/src/test/scala/integration/kafka/tools/MaintenanceBrokerTestUtils.scala new file mode 100644 index 0000000000000..82edae2829139 --- /dev/null +++ b/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) + } + +} diff --git a/core/src/test/scala/unit/kafka/server/MaintenanceBrokerTest.scala b/core/src/test/scala/unit/kafka/server/MaintenanceBrokerTest.scala index c97740136623b..73d4df8d9a2f4 100644 --- a/core/src/test/scala/unit/kafka/server/MaintenanceBrokerTest.scala +++ b/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) } }