AndrewJSchofield commented on code in PR #19493:
URL: https://github.com/apache/kafka/pull/19493#discussion_r2083608078
##########
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:
Let's have a test for an invalid resource type.
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -2874,16 +2874,55 @@ class KafkaApis(val requestChannel: RequestChannel,
}
}
- def handleListClientMetricsResources(request: RequestChannel.Request): Unit
= {
- val listClientMetricsResourcesRequest =
request.body[ListClientMetricsResourcesRequest]
+ private def handleListConfigResources(request: RequestChannel.Request): Unit
= {
+ val listConfigResourcesRequest = request.body[ListConfigResourcesRequest]
if (!authHelper.authorize(request.context, DESCRIBE_CONFIGS, CLUSTER,
CLUSTER_NAME)) {
- requestHelper.sendMaybeThrottle(request,
listClientMetricsResourcesRequest.getErrorResponse(Errors.CLUSTER_AUTHORIZATION_FAILED.exception))
+ requestHelper.sendMaybeThrottle(request,
listConfigResourcesRequest.getErrorResponse(Errors.CLUSTER_AUTHORIZATION_FAILED.exception))
} else {
- val data = new
ListClientMetricsResourcesResponseData().setClientMetricsResources(
- clientMetricsManager.listClientMetricsResources.stream.map(
- name => new
ClientMetricsResource().setName(name)).collect(Collectors.toList()))
- requestHelper.sendMaybeThrottle(request, new
ListClientMetricsResourcesResponse(data))
+ val data = new ListConfigResourcesResponseData()
+
+ if (listConfigResourcesRequest.onlySupportClientMetrics) {
+ // Version 0 only supports client metrics.
Review Comment:
I think this version-specific code would be better inside
`ListConfigResourceRequest.java`. You could add `resourceTypes()` and
initialize the value to a singleton set containing
`ConfigResource.Type.CLIENT_METRICS.id` if the request was v0. wdyt?
##########
clients/src/main/java/org/apache/kafka/common/requests/ListConfigResourcesRequest.java:
##########
@@ -72,4 +72,7 @@ public String toString(boolean verbose) {
return data.toString();
}
+ public boolean onlySupportClientMetrics() {
Review Comment:
nit: I think `onlySupportsClientMetrics` would be preferable.
--
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
