This is an automated email from the ASF dual-hosted git repository.
cmccabe pushed a commit to branch 2.8
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/2.8 by this push:
new 6eda939 KAFKA-12235: Fix a bug where describeConfigs does not work on
2+ config keys (#9990)
6eda939 is described below
commit 6eda9398f01a186675af3e261cf485ed3f0ba670
Author: Ivan Yurchenko <[email protected]>
AuthorDate: Fri Feb 26 21:33:21 2021 +0200
KAFKA-12235: Fix a bug where describeConfigs does not work on 2+ config
keys (#9990)
Fix a bug where if more than one configuration key is supplied,
describeConfigs fails to return any results at all.
Reviewers: Colin P. McCabe <[email protected]>
---
core/src/main/scala/kafka/server/ConfigHelper.scala | 4 ++--
.../unit/kafka/server/ZkAdminManagerTest.scala | 21 +++++++++++++++++++++
2 files changed, 23 insertions(+), 2 deletions(-)
diff --git a/core/src/main/scala/kafka/server/ConfigHelper.scala
b/core/src/main/scala/kafka/server/ConfigHelper.scala
index 2bc8dd9..779bea8 100644
--- a/core/src/main/scala/kafka/server/ConfigHelper.scala
+++ b/core/src/main/scala/kafka/server/ConfigHelper.scala
@@ -47,11 +47,11 @@ class ConfigHelper(metadataCache: MetadataCache, config:
KafkaConfig, configRepo
def createResponseConfig(configs: Map[String, Any],
createConfigEntry: (String, Any) =>
DescribeConfigsResponseData.DescribeConfigsResourceResult):
DescribeConfigsResponseData.DescribeConfigsResult = {
- val filteredConfigPairs = if (resource.configurationKeys == null)
+ val filteredConfigPairs = if (resource.configurationKeys == null ||
resource.configurationKeys.isEmpty)
configs.toBuffer
else
configs.filter { case (configName, _) =>
- resource.configurationKeys.asScala.forall(_.contains(configName))
+ resource.configurationKeys.asScala.contains(configName)
}.toBuffer
val configEntries = filteredConfigPairs.map { case (name, value) =>
createConfigEntry(name, value) }
diff --git a/core/src/test/scala/unit/kafka/server/ZkAdminManagerTest.scala
b/core/src/test/scala/unit/kafka/server/ZkAdminManagerTest.scala
index ae0ac79..3533133 100644
--- a/core/src/test/scala/unit/kafka/server/ZkAdminManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ZkAdminManagerTest.scala
@@ -34,6 +34,8 @@ import java.util.Properties
import kafka.server.metadata.ZkConfigRepository
+import scala.jdk.CollectionConverters._
+
class ZkAdminManagerTest {
private val zkClient: KafkaZkClient =
EasyMock.createNiceMock(classOf[KafkaZkClient])
@@ -86,6 +88,25 @@ class ZkAdminManagerTest {
}
@Test
+ def testDescribeConfigsWithConfigurationKeys(): Unit = {
+ EasyMock.expect(zkClient.getEntityConfigs(ConfigType.Topic,
topic)).andReturn(TestUtils.createBrokerConfig(brokerId, "zk"))
+ EasyMock.expect(metadataCache.contains(topic)).andReturn(true)
+
+ EasyMock.replay(zkClient, metadataCache)
+
+ val resources = List(new
DescribeConfigsRequestData.DescribeConfigsResource()
+ .setResourceName(topic)
+ .setResourceType(ConfigResource.Type.TOPIC.id)
+ .setConfigurationKeys(List("retention.ms", "retention.bytes",
"segment.bytes").asJava)
+ )
+ val configHelper = createConfigHelper(metadataCache, zkClient)
+ val results: List[DescribeConfigsResponseData.DescribeConfigsResult] =
configHelper.describeConfigs(resources, true, true)
+ assertEquals(Errors.NONE.code, results.head.errorCode())
+ val resultConfigKeys = results.head.configs().asScala.map(r =>
r.name()).toSet
+ assertEquals(Set("retention.ms", "retention.bytes", "segment.bytes"),
resultConfigKeys)
+ }
+
+ @Test
def testDescribeConfigsWithDocumentation(): Unit = {
EasyMock.expect(zkClient.getEntityConfigs(ConfigType.Topic,
topic)).andReturn(new Properties)
EasyMock.expect(zkClient.getEntityConfigs(ConfigType.Broker,
brokerId.toString)).andReturn(new Properties)
