FrankYang0529 commented on code in PR #19493:
URL: https://github.com/apache/kafka/pull/19493#discussion_r2086922412
##########
core/src/test/scala/unit/kafka/server/KafkaApisTest.scala:
##########
@@ -10929,47 +10928,231 @@ class KafkaApisTest extends Logging {
}
@Test
- def testListClientMetricsResources(): Unit = {
- val request = buildRequest(new
ListClientMetricsResourcesRequest.Builder(new
ListClientMetricsResourcesRequestData()).build())
- metadataCache = new KRaftMetadataCache(brokerId, () =>
KRaftVersion.KRAFT_VERSION_0)
+ def testListConfigResourcesV0(): Unit = {
+ val request = buildRequest(new ListConfigResourcesRequest.Builder(new
ListConfigResourcesRequestData()).build(0))
+ metadataCache = mock(classOf[KRaftMetadataCache])
+
+ val resources = util.Set.of("client-metric1", "client-metric2")
+ when(clientMetricsManager.listClientMetricsResources).thenReturn(resources)
- val resources = new mutable.HashSet[String]
- resources.add("test1")
- resources.add("test2")
-
when(clientMetricsManager.listClientMetricsResources).thenReturn(resources.asJava)
kafkaApis = createKafkaApis()
kafkaApis.handle(request, RequestLocal.noCaching)
- val response =
verifyNoThrottling[ListClientMetricsResourcesResponse](request)
- val expectedResponse = new
ListClientMetricsResourcesResponseData().setClientMetricsResources(
- resources.map(resource => new
ClientMetricsResource().setName(resource)).toBuffer.asJava)
- assertEquals(expectedResponse, response.data)
+ val response = verifyNoThrottling[ListConfigResourcesResponse](request)
+ val expectedResponse = new ListConfigResourcesResponse(
+ new ListConfigResourcesResponseData()
+ .setConfigResources(
+ resources.stream.map(resource =>
+ new
ListConfigResourcesResponseData.ConfigResource().setResourceName(resource)
+ ).toList)
+ )
+ assertEquals(expectedResponse.data, response.data)
+
+ verify(metadataCache, never).getAllTopics
+ verify(groupConfigManager, never).groupIds
+ verify(metadataCache, never).getBrokerNodes(any)
}
@Test
- def testListClientMetricsResourcesEmptyResponse(): Unit = {
- val request = buildRequest(new
ListClientMetricsResourcesRequest.Builder(new
ListClientMetricsResourcesRequestData()).build())
- metadataCache = new KRaftMetadataCache(brokerId, () =>
KRaftVersion.KRAFT_VERSION_0)
+ def testListConfigResourcesV1WithEmptyResourceTypes(): Unit = {
+ val request = buildRequest(new ListConfigResourcesRequest.Builder(new
ListConfigResourcesRequestData()).build(1))
+ metadataCache = mock(classOf[KRaftMetadataCache])
+
+ val clientMetrics = util.Set.of("client-metric1", "client-metric2")
+ val topics = util.Set.of("topic1", "topic2")
+ val groupIds = util.List.of("group1", "group2")
+ val nodeIds = util.List.of(1, 2)
+
when(clientMetricsManager.listClientMetricsResources).thenReturn(clientMetrics)
+ when(metadataCache.getAllTopics).thenReturn(topics)
+ when(groupConfigManager.groupIds).thenReturn(groupIds)
+
when(metadataCache.getBrokerNodes(any())).thenReturn(nodeIds.stream().map(id =>
new Node(id, "localhost", 1234)).toList)
+
+ kafkaApis = createKafkaApis()
+ kafkaApis.handle(request, RequestLocal.noCaching)
+ val response = verifyNoThrottling[ListConfigResourcesResponse](request)
+ val expectedResponse = new ListConfigResourcesResponse(
+ new ListConfigResourcesResponseData()
+ .setConfigResources(
+ util.stream.Stream.of(
+ groupIds.stream().map(resource =>
+ new
ListConfigResourcesResponseData.ConfigResource().setResourceName(resource).setResourceType(ConfigResource.Type.GROUP.id)
+ ).toList,
+ clientMetrics.stream.map(resource =>
+ new
ListConfigResourcesResponseData.ConfigResource().setResourceName(resource).setResourceType(ConfigResource.Type.CLIENT_METRICS.id)
+ ).toList,
+ nodeIds.stream().map(resource =>
+ new
ListConfigResourcesResponseData.ConfigResource().setResourceName(resource.toString).setResourceType(ConfigResource.Type.BROKER_LOGGER.id)
+ ).toList,
+ nodeIds.stream().map(resource =>
+ new
ListConfigResourcesResponseData.ConfigResource().setResourceName(resource.toString).setResourceType(ConfigResource.Type.BROKER.id)
+ ).toList,
+ topics.stream().map(resource =>
+ new
ListConfigResourcesResponseData.ConfigResource().setResourceName(resource).setResourceType(ConfigResource.Type.TOPIC.id)
+ ).toList
+ ).flatMap(s => s.stream).toList)
+ )
+ assertEquals(expectedResponse.data, response.data)
+ }
+
+ @Test
+ def testListConfigResourcesV1WithGroup(): Unit = {
+ val request = buildRequest(new ListConfigResourcesRequest.Builder(new
ListConfigResourcesRequestData()
+ .setResourceTypes(util.List.of(ConfigResource.Type.GROUP.id))).build(1))
+ metadataCache = mock(classOf[KRaftMetadataCache])
+
+ val groupIds = util.List.of("group1", "group2")
+ when(groupConfigManager.groupIds).thenReturn(groupIds)
+
+ kafkaApis = createKafkaApis()
+ kafkaApis.handle(request, RequestLocal.noCaching)
+ val response = verifyNoThrottling[ListConfigResourcesResponse](request)
+ val expectedResponse = new ListConfigResourcesResponse(
+ new ListConfigResourcesResponseData()
+ .setConfigResources(
+ groupIds.stream().map(resource =>
+ new
ListConfigResourcesResponseData.ConfigResource().setResourceName(resource).setResourceType(ConfigResource.Type.GROUP.id)
+ ).toList,
+ )
+ )
+ assertEquals(expectedResponse.data, response.data)
+
+ verify(metadataCache, never).getAllTopics
+ verify(clientMetricsManager, never).listClientMetricsResources
+ verify(metadataCache, never).getBrokerNodes(any)
+ }
+
+ @Test
+ def testListConfigResourcesV1WithClientMetrics(): Unit = {
+ val request = buildRequest(new ListConfigResourcesRequest.Builder(new
ListConfigResourcesRequestData()
+
.setResourceTypes(util.List.of(ConfigResource.Type.CLIENT_METRICS.id))).build(1))
+ metadataCache = mock(classOf[KRaftMetadataCache])
+
+ val clientMetrics = util.Set.of("client-metric1", "client-metric2")
+
when(clientMetricsManager.listClientMetricsResources).thenReturn(clientMetrics)
+
+ kafkaApis = createKafkaApis()
+ kafkaApis.handle(request, RequestLocal.noCaching)
+ val response = verifyNoThrottling[ListConfigResourcesResponse](request)
+ val expectedResponse = new ListConfigResourcesResponse(
+ new ListConfigResourcesResponseData()
+ .setConfigResources(
+ clientMetrics.stream.map(resource =>
+ new
ListConfigResourcesResponseData.ConfigResource().setResourceName(resource).setResourceType(ConfigResource.Type.CLIENT_METRICS.id)
+ ).toList
+ )
+ )
+ assertEquals(expectedResponse.data, response.data)
+
+ verify(metadataCache, never).getAllTopics
+ verify(groupConfigManager, never).groupIds
+ verify(metadataCache, never).getBrokerNodes(any)
+ }
+
+ @Test
+ def testListConfigResourcesV1WithBrokerLogger(): Unit = {
+ val request = buildRequest(new ListConfigResourcesRequest.Builder(new
ListConfigResourcesRequestData()
+
.setResourceTypes(util.List.of(ConfigResource.Type.BROKER_LOGGER.id))).build(1))
+ metadataCache = mock(classOf[KRaftMetadataCache])
+
+ val nodeIds = util.List.of(1, 2)
+
when(metadataCache.getBrokerNodes(any())).thenReturn(nodeIds.stream().map(id =>
new Node(id, "localhost", 1234)).toList)
+
+ kafkaApis = createKafkaApis()
+ kafkaApis.handle(request, RequestLocal.noCaching)
+ val response = verifyNoThrottling[ListConfigResourcesResponse](request)
+ val expectedResponse = new ListConfigResourcesResponse(
+ new ListConfigResourcesResponseData()
+ .setConfigResources(
+ nodeIds.stream().map(resource =>
+ new
ListConfigResourcesResponseData.ConfigResource().setResourceName(resource.toString).setResourceType(ConfigResource.Type.BROKER_LOGGER.id)
+ ).toList)
+ )
+ assertEquals(expectedResponse.data, response.data)
+
+ verify(metadataCache, never).getAllTopics
+ verify(groupConfigManager, never).groupIds
+ verify(clientMetricsManager, never).listClientMetricsResources
+ }
+
+ @Test
+ def testListConfigResourcesV1WithBroker(): Unit = {
+ val request = buildRequest(new ListConfigResourcesRequest.Builder(new
ListConfigResourcesRequestData()
+ .setResourceTypes(util.List.of(ConfigResource.Type.BROKER.id))).build(1))
+ metadataCache = mock(classOf[KRaftMetadataCache])
+
+ val nodeIds = util.List.of(1, 2)
+
when(metadataCache.getBrokerNodes(any())).thenReturn(nodeIds.stream().map(id =>
new Node(id, "localhost", 1234)).toList)
+
+ kafkaApis = createKafkaApis()
+ kafkaApis.handle(request, RequestLocal.noCaching)
+ val response = verifyNoThrottling[ListConfigResourcesResponse](request)
+ val expectedResponse = new ListConfigResourcesResponse(
+ new ListConfigResourcesResponseData()
+ .setConfigResources(
+ nodeIds.stream().map(resource =>
+ new
ListConfigResourcesResponseData.ConfigResource().setResourceName(resource.toString).setResourceType(ConfigResource.Type.BROKER.id)
+ ).toList)
+ )
+ assertEquals(expectedResponse.data, response.data)
+
+ verify(metadataCache, never).getAllTopics
+ verify(groupConfigManager, never).groupIds
+ verify(clientMetricsManager, never).listClientMetricsResources
+ }
+
+ @Test
+ def testListConfigResourcesV1WithTopic(): Unit = {
+ val request = buildRequest(new ListConfigResourcesRequest.Builder(new
ListConfigResourcesRequestData()
+ .setResourceTypes(util.List.of(ConfigResource.Type.TOPIC.id))).build(1))
Review Comment:
Add a new case `testListConfigResourcesV1WithUnknown`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
