KAFKA-10557: Missing docs when describing topic configs including (#9360)

Reviewers: Rajini Sivaram <[email protected]>

Co-authored-by: Edoardo Comar <[email protected]>
Co-authored-by: Mickael Maison <[email protected]>

Author: edoardocomar

core/src/main/scala/kafka/server/AdminManager.scala

@@ -764,7 +764,7 @@ class AdminManager(val config: KafkaConfig,
     val source = if (allSynonyms.isEmpty) ConfigSource.DEFAULT_CONFIG.id else allSynonyms.head.source
     val synonyms = if (!includeSynonyms) List.empty else allSynonyms
     val dataType = configResponseType(configEntryType)
-    val configDocumentation = if (includeDocumentation) brokerDocumentation(name) else null
+    val configDocumentation = if (includeDocumentation) logConfig.documentationOf(name) else null
     new DescribeConfigsResponseData.DescribeConfigsResourceResult()
       .setName(name).setValue(valueAsString).setConfigSource(source)
       .setIsSensitive(isSensitive).setReadOnly(false).setSynonyms(synonyms.asJava)

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

@@ -29,6 +29,9 @@ import org.apache.kafka.common.protocol.Errors
 import org.junit.{After, Test}
 import org.junit.Assert.assertEquals
 import org.junit.Assert.assertFalse
+import org.junit.Assert.assertNotNull
+import org.junit.Assert.assertNotEquals
+import java.util.Properties
 
 class AdminManagerTest {
 
@@ -80,4 +83,33 @@ class AdminManagerTest {
     assertEquals(Errors.NONE.code, results.head.errorCode())
     assertFalse("Should return configs", results.head.configs().isEmpty)
   }
+
+  @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)
+    EasyMock.expect(metadataCache.contains(topic)).andReturn(true)
+    EasyMock.replay(zkClient, metadataCache)
+
+    val adminManager = createAdminManager()
+
+    val resources = List(
+      new DescribeConfigsRequestData.DescribeConfigsResource()
+        .setResourceName(topic)
+        .setResourceType(ConfigResource.Type.TOPIC.id),
+      new DescribeConfigsRequestData.DescribeConfigsResource()
+        .setResourceName(brokerId.toString)
+        .setResourceType(ConfigResource.Type.BROKER.id))
+
+    val results: List[DescribeConfigsResponseData.DescribeConfigsResult] = adminManager.describeConfigs(resources, true, true)
+    assertEquals(2, results.size)
+    results.foreach(r => {
+      assertEquals(Errors.NONE.code, r.errorCode)
+      assertFalse("Should return configs", r.configs.isEmpty)
+      r.configs.forEach(c => {
+        assertNotNull(s"Config ${c.name} should have non null documentation", c.documentation)
+        assertNotEquals(s"Config ${c.name} should have non blank documentation", "", c.documentation.trim)
+      })
+    })
+  }
 }