apoorvmittal10 commented on code in PR #14632:
URL: https://github.com/apache/kafka/pull/14632#discussion_r1387101615
##########
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:
I have added alterResult as well, thanks. `all()` is need as it ensures the
verification of returned `future` is executed.
--
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
