junrao commented on code in PR #14632:
URL: https://github.com/apache/kafka/pull/14632#discussion_r1385792007
##########
core/src/main/scala/kafka/server/DynamicConfig.scala:
##########
@@ -111,6 +112,16 @@ object DynamicConfig {
}
}
+ object ClientMetrics {
+ private val clientConfigs = kafka.metrics.ClientMetricsConfigs.configDef()
+
+ def configKeys: util.Map[String, ConfigDef.ConfigKey] =
clientConfigs.configKeys
+
+ def names: util.Set[String] = clientConfigs.names
+
+ def validate(props: Properties) = DynamicConfig.validate(clientConfigs,
props, customPropsAllowed = false)
Review Comment:
This is for ZK-based controller. Since ClientMetrics is not supported there,
do we need this? Ditto for `configKeys`.
##########
core/src/main/scala/kafka/admin/ConfigCommand.scala:
##########
@@ -536,6 +552,8 @@ object ConfigCommand extends Logging {
adminClient.listTopics(new
ListTopicsOptions().listInternal(true)).names().get().asScala.toSeq
case ConfigType.Broker | BrokerLoggerConfigType =>
adminClient.describeCluster(new
DescribeClusterOptions()).nodes().get().asScala.map(_.idString).toSeq :+
BrokerDefaultEntityName
+ case ConfigType.ClientMetrics =>
+ throw new InvalidRequestException("Client metrics entity-name is
required")
Review Comment:
I guess this will be improved through
https://cwiki.apache.org/confluence/display/KAFKA/KIP-1000%3A+List+Client+Metrics+Configuration+Resources
?
##########
core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala:
##########
@@ -1628,6 +1628,122 @@ class ConfigCommandTest extends Logging {
Seq("<default>/clients/client-3", sanitizedPrincipal +
"/clients/client-2"))
}
+ @Test
+ def shouldAlterClientMetricsConfig(): Unit = {
+ val node = new Node(1, "localhost", 9092)
+ verifyAlterClientMetricsConfig(node, "1", List("--entity-name", "1"))
+ }
+
+ def verifyAlterClientMetricsConfig(node: Node, resourceName: String,
resourceOpts: List[String]): Unit = {
Review Comment:
Could this be private?
##########
core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala:
##########
@@ -1628,6 +1628,122 @@ class ConfigCommandTest extends Logging {
Seq("<default>/clients/client-3", sanitizedPrincipal +
"/clients/client-2"))
}
+ @Test
+ def shouldAlterClientMetricsConfig(): Unit = {
+ val node = new Node(1, "localhost", 9092)
+ verifyAlterClientMetricsConfig(node, "1", List("--entity-name", "1"))
+ }
+
+ def verifyAlterClientMetricsConfig(node: Node, resourceName: String,
resourceOpts: List[String]): Unit = {
+ val optsList = List("--bootstrap-server", "localhost:9092",
+ "--entity-type", "client-metrics",
+ "--alter",
+ "--delete-config", "interval.ms",
+ "--add-config", "metrics=org.apache.kafka.consumer.," +
+
"match=[client_software_name=kafka.python,client_software_version=1\\.2\\..*]")
++ resourceOpts
+ val alterOpts = new ConfigCommandOptions(optsList.toArray)
+
+ val resource = new ConfigResource(ConfigResource.Type.CLIENT_METRICS,
resourceName)
+ val configEntries = util.Collections.singletonList(new
ConfigEntry("interval.ms", "1000",
+ ConfigEntry.ConfigSource.DYNAMIC_CLIENT_METRICS_CONFIG, false, false,
Collections.emptyList,
+ ConfigEntry.ConfigType.UNKNOWN, null))
+ val future = new KafkaFutureImpl[util.Map[ConfigResource, Config]]
+ future.complete(util.Collections.singletonMap(resource, new
Config(configEntries)))
+ val describeResult: DescribeConfigsResult =
mock(classOf[DescribeConfigsResult])
+ when(describeResult.all()).thenReturn(future)
+
+ val alterFuture = new KafkaFutureImpl[Void]
+ alterFuture.complete(null)
+ val alterResult: AlterConfigsResult = mock(classOf[AlterConfigsResult])
+ when(alterResult.all()).thenReturn(alterFuture)
+
+ val mockAdminClient = new
MockAdminClient(util.Collections.singletonList(node), node) {
+ override def describeConfigs(resources: util.Collection[ConfigResource],
options: DescribeConfigsOptions): DescribeConfigsResult = {
+ assertFalse(options.includeSynonyms(), "Config synonyms requested
unnecessarily")
+ assertEquals(1, resources.size)
+ val resource = resources.iterator.next
+ assertEquals(ConfigResource.Type.CLIENT_METRICS, resource.`type`)
+ assertEquals(resourceName, resource.name)
+ describeResult
+ }
+
+ override def incrementalAlterConfigs(configs: util.Map[ConfigResource,
util.Collection[AlterConfigOp]], options: AlterConfigsOptions):
AlterConfigsResult = {
+ assertEquals(1, configs.size)
+ val entry = configs.entrySet.iterator.next
+ val resource = entry.getKey
+ val alterConfigOps = entry.getValue
+ assertEquals(ConfigResource.Type.CLIENT_METRICS, resource.`type`)
+ assertEquals(3, alterConfigOps.size)
+
+ val expectedConfigOps = List(
+ new AlterConfigOp(new ConfigEntry("match",
"client_software_name=kafka.python,client_software_version=1\\.2\\..*"),
AlterConfigOp.OpType.SET),
+ new AlterConfigOp(new ConfigEntry("metrics",
"org.apache.kafka.consumer."), AlterConfigOp.OpType.SET),
+ new AlterConfigOp(new ConfigEntry("interval.ms", ""),
AlterConfigOp.OpType.DELETE)
+ )
+ assertEquals(expectedConfigOps, alterConfigOps.asScala.toList)
+ alterResult
+ }
+ }
+ ConfigCommand.alterConfig(mockAdminClient, alterOpts)
+ verify(describeResult).all()
Review Comment:
Do we need to verify `alterResult` too? Also, why do we need to call `all()`?
--
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
