[kafka] branch 2.7 updated: KAFKA-12235: Fix a bug where describeConfigs does not work on 2+ config keys (#9990)

[cm...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

cmccabe pushed a commit to branch 2.7
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/2.7 by this push:
     new dcbc62d  KAFKA-12235: Fix a bug where describeConfigs does not work on 2+ config keys (#9990)
dcbc62d is described below

commit dcbc62df89d783d9cc19e21453cf7264a623cf8b
Author: Ivan Yurchenko <iv...@aiven.io>
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 <cm...@apache.org>
    
    Conflicts:
    	core/src/main/scala/kafka/server/ConfigHelper.scala
    	core/src/test/scala/unit/kafka/server/AdminManagerTest.scala
---
 core/src/main/scala/kafka/server/AdminManager.scala |  4 ++--
 .../scala/unit/kafka/server/AdminManagerTest.scala  | 21 +++++++++++++++++++++
 2 files changed, 23 insertions(+), 2 deletions(-)

diff --git a/core/src/main/scala/kafka/server/AdminManager.scala b/core/src/main/scala/kafka/server/AdminManager.scala
index 841efd5..8c048fd 100644
--- a/core/src/main/scala/kafka/server/AdminManager.scala
+++ b/core/src/main/scala/kafka/server/AdminManager.scala
@@ -390,11 +390,11 @@ class AdminManager(val config: KafkaConfig,
       }
       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/AdminManagerTest.scala b/core/src/test/scala/unit/kafka/server/AdminManagerTest.scala
index e97162f..e6210e8 100644
--- a/core/src/test/scala/unit/kafka/server/AdminManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/server/AdminManagerTest.scala
@@ -33,6 +33,8 @@ import org.junit.Assert.assertNotNull
 import org.junit.Assert.assertNotEquals
 import java.util.Properties
 
+import scala.jdk.CollectionConverters._
+
 class AdminManagerTest {
 
   private val zkClient: KafkaZkClient = EasyMock.createNiceMock(classOf[KafkaZkClient])
@@ -85,6 +87,25 @@ class AdminManagerTest {
   }
 
   @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 adminManager = createAdminManager()
+    val results: List[DescribeConfigsResponseData.DescribeConfigsResult] = adminManager.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)