You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ju...@apache.org on 2023/11/28 17:24:35 UTC
(kafka) branch trunk updated: KAFKA-15681: Add support of client-metrics in kafka-configs.sh (KIP-714) (#14632)
This is an automated email from the ASF dual-hosted git repository.
junrao pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 38f2faf83f7 KAFKA-15681: Add support of client-metrics in kafka-configs.sh (KIP-714) (#14632)
38f2faf83f7 is described below
commit 38f2faf83f7e83823767aa71b8fd0850dae93b6e
Author: Apoorv Mittal <am...@confluent.io>
AuthorDate: Tue Nov 28 22:54:25 2023 +0530
KAFKA-15681: Add support of client-metrics in kafka-configs.sh (KIP-714) (#14632)
The PR adds support of alter/describe configs for client-metrics as defined in KIP-714
Reviewers: Andrew Schofield <as...@confluent.io>, Jun Rao <ju...@gmail.com>
---
checkstyle/suppressions.xml | 2 +-
.../kafka/clients/admin/MockAdminClient.java | 37 +++++++
.../src/main/scala/kafka/admin/ConfigCommand.scala | 38 +++++--
.../main/scala/kafka/server/DynamicConfig.scala | 7 ++
.../main/scala/kafka/server/ZkConfigManager.scala | 3 +-
.../kafka/server/metadata/ZkConfigRepository.scala | 2 +
.../scala/unit/kafka/admin/ConfigCommandTest.scala | 117 +++++++++++++++++++++
.../server/metadata/ZkConfigRepositoryTest.scala | 5 +-
8 files changed, 200 insertions(+), 11 deletions(-)
diff --git a/checkstyle/suppressions.xml b/checkstyle/suppressions.xml
index cd6ac4a13dc..bcdfede03f3 100644
--- a/checkstyle/suppressions.xml
+++ b/checkstyle/suppressions.xml
@@ -89,7 +89,7 @@
files="(Utils|Topic|KafkaLZ4BlockOutputStream|AclData|JoinGroupRequest).java"/>
<suppress checks="CyclomaticComplexity"
- files="(AbstractFetch|ConsumerCoordinator|FetchCollector|OffsetFetcherUtils|KafkaProducer|Sender|ConfigDef|KerberosLogin|AbstractRequest|AbstractResponse|Selector|SslFactory|SslTransportLayer|SaslClientAuthenticator|SaslClientCallbackHandler|SaslServerAuthenticator|AbstractCoordinator|TransactionManager|AbstractStickyAssignor|DefaultSslEngineFactory|Authorizer|RecordAccumulator|MemoryRecords|FetchSessionHandler).java"/>
+ files="(AbstractFetch|ConsumerCoordinator|FetchCollector|OffsetFetcherUtils|KafkaProducer|Sender|ConfigDef|KerberosLogin|AbstractRequest|AbstractResponse|Selector|SslFactory|SslTransportLayer|SaslClientAuthenticator|SaslClientCallbackHandler|SaslServerAuthenticator|AbstractCoordinator|TransactionManager|AbstractStickyAssignor|DefaultSslEngineFactory|Authorizer|RecordAccumulator|MemoryRecords|FetchSessionHandler|MockAdminClient).java"/>
<suppress checks="JavaNCSS"
files="(AbstractRequest|AbstractResponse|KerberosLogin|WorkerSinkTaskTest|TransactionManagerTest|SenderTest|KafkaAdminClient|ConsumerCoordinatorTest|KafkaAdminClientTest|KafkaRaftClientTest).java"/>
diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClient.java b/clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClient.java
index 515b02e6965..52188c68df4 100644
--- a/clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClient.java
+++ b/clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClient.java
@@ -90,6 +90,7 @@ public class MockAdminClient extends AdminClient {
private final String clusterId;
private final List<List<String>> brokerLogDirs;
private final List<Map<String, String>> brokerConfigs;
+ private final Map<String, Map<String, String>> clientMetricsConfigs;
private Node controller;
private int timeoutNextRequests = 0;
@@ -238,6 +239,7 @@ public class MockAdminClient extends AdminClient {
this.defaultReplicationFactor = defaultReplicationFactor;
this.brokerLogDirs = brokerLogDirs;
this.brokerConfigs = new ArrayList<>();
+ this.clientMetricsConfigs = new HashMap<>();
for (int i = 0; i < brokers.size(); i++) {
final Map<String, String> config = new HashMap<>();
config.put("default.replication.factor", String.valueOf(defaultReplicationFactor));
@@ -823,6 +825,13 @@ public class MockAdminClient extends AdminClient {
}
throw new UnknownTopicOrPartitionException("Resource " + resource + " not found.");
}
+ case CLIENT_METRICS: {
+ String resourceName = resource.name();
+ if (resourceName.isEmpty()) {
+ throw new InvalidRequestException("Empty resource name");
+ }
+ return toConfigObject(clientMetricsConfigs.get(resourceName));
+ }
default:
throw new UnsupportedOperationException("Not implemented yet");
}
@@ -916,6 +925,34 @@ public class MockAdminClient extends AdminClient {
topicMetadata.configs = newMap;
return null;
}
+ case CLIENT_METRICS: {
+ String resourceName = resource.name();
+
+ if (resourceName.isEmpty()) {
+ return new InvalidRequestException("Empty resource name");
+ }
+
+ if (!clientMetricsConfigs.containsKey(resourceName)) {
+ clientMetricsConfigs.put(resourceName, new HashMap<>());
+ }
+
+ HashMap<String, String> newMap = new HashMap<>(clientMetricsConfigs.get(resourceName));
+ for (AlterConfigOp op : ops) {
+ switch (op.opType()) {
+ case SET:
+ newMap.put(op.configEntry().name(), op.configEntry().value());
+ break;
+ case DELETE:
+ newMap.remove(op.configEntry().name());
+ break;
+ default:
+ return new InvalidRequestException(
+ "Unsupported op type " + op.opType());
+ }
+ }
+ clientMetricsConfigs.put(resourceName, newMap);
+ return null;
+ }
default:
return new UnsupportedOperationException();
}
diff --git a/core/src/main/scala/kafka/admin/ConfigCommand.scala b/core/src/main/scala/kafka/admin/ConfigCommand.scala
index 25d400918f6..37d41458c39 100644
--- a/core/src/main/scala/kafka/admin/ConfigCommand.scala
+++ b/core/src/main/scala/kafka/admin/ConfigCommand.scala
@@ -29,7 +29,7 @@ import kafka.zk.{AdminZkClient, KafkaZkClient}
import org.apache.kafka.clients.admin.{Admin, AlterClientQuotasOptions, AlterConfigOp, AlterConfigsOptions, ConfigEntry, DescribeClusterOptions, DescribeConfigsOptions, ListTopicsOptions, ScramCredentialInfo, UserScramCredentialDeletion, UserScramCredentialUpsertion, Config => JConfig, ScramMechanism => PublicScramMechanism}
import org.apache.kafka.common.config.{ConfigResource, TopicConfig}
import org.apache.kafka.common.config.types.Password
-import org.apache.kafka.common.errors.InvalidConfigurationException
+import org.apache.kafka.common.errors.{InvalidConfigurationException, InvalidRequestException}
import org.apache.kafka.common.internals.Topic
import org.apache.kafka.common.quota.{ClientQuotaAlteration, ClientQuotaEntity, ClientQuotaFilter, ClientQuotaFilterComponent}
import org.apache.kafka.common.security.JaasUtils
@@ -44,7 +44,7 @@ import scala.jdk.CollectionConverters._
import scala.collection._
/**
- * This script can be used to change configs for topics/clients/users/brokers/ips dynamically
+ * This script can be used to change configs for topics/clients/users/brokers/ips/client-metrics dynamically
* An entity described or altered by the command may be one of:
* <ul>
* <li> topic: --topic <topic> OR --entity-type topics --entity-name <topic>
@@ -55,6 +55,7 @@ import scala.collection._
* <li> broker: --broker <broker-id> OR --entity-type brokers --entity-name <broker-id>
* <li> broker-logger: --broker-logger <broker-id> OR --entity-type broker-loggers --entity-name <broker-id>
* <li> ip: --ip <ip> OR --entity-type ips --entity-name <ip>
+ * <li> client-metrics: --client-metrics <name> OR --entity-type client-metrics --entity-name <name>
* </ul>
* --entity-type <users|clients|brokers|ips> --entity-default may be specified in place of --entity-type <users|clients|brokers|ips> --entity-name <entityName>
* when describing or altering default configuration for users, clients, brokers, or ips, respectively.
@@ -76,7 +77,7 @@ object ConfigCommand extends Logging {
val BrokerDefaultEntityName = ""
val BrokerLoggerConfigType = "broker-loggers"
- val BrokerSupportedConfigTypes = ConfigType.all :+ BrokerLoggerConfigType
+ val BrokerSupportedConfigTypes = ConfigType.all :+ BrokerLoggerConfigType :+ ConfigType.ClientMetrics
val ZkSupportedConfigTypes = Seq(ConfigType.User, ConfigType.Broker)
val DefaultScramIterations = 4096
@@ -84,7 +85,7 @@ object ConfigCommand extends Logging {
try {
val opts = new ConfigCommandOptions(args)
- CommandLineUtils.maybePrintHelpOrVersion(opts, "This tool helps to manipulate and describe entity config for a topic, client, user, broker or ip")
+ CommandLineUtils.maybePrintHelpOrVersion(opts, "This tool helps to manipulate and describe entity config for a topic, client, user, broker, ip or client-metrics")
opts.checkArgs()
@@ -445,6 +446,21 @@ object ConfigCommand extends Logging {
if (unknownConfigs.nonEmpty)
throw new IllegalArgumentException(s"Only connection quota configs can be added for '${ConfigType.Ip}' using --bootstrap-server. Unexpected config names: ${unknownConfigs.mkString(",")}")
alterQuotaConfigs(adminClient, entityTypes, entityNames, configsToBeAddedMap, configsToBeDeleted)
+ case ConfigType.ClientMetrics =>
+ val oldConfig = getResourceConfig(adminClient, entityTypeHead, entityNameHead, includeSynonyms = false, describeAll = false)
+ .map { entry => (entry.name, entry) }.toMap
+
+ // fail the command if any of the configs to be deleted does not exist
+ val invalidConfigs = configsToBeDeleted.filterNot(oldConfig.contains)
+ if (invalidConfigs.nonEmpty)
+ throw new InvalidConfigurationException(s"Invalid config(s): ${invalidConfigs.mkString(",")}")
+
+ val configResource = new ConfigResource(ConfigResource.Type.CLIENT_METRICS, entityNameHead)
+ val alterOptions = new AlterConfigsOptions().timeoutMs(30000).validateOnly(false)
+ val alterEntries = (configsToBeAdded.values.map(new AlterConfigOp(_, AlterConfigOp.OpType.SET))
+ ++ configsToBeDeleted.map { k => new AlterConfigOp(new ConfigEntry(k, ""), AlterConfigOp.OpType.DELETE) }
+ ).asJavaCollection
+ adminClient.incrementalAlterConfigs(Map(configResource -> alterEntries).asJava, alterOptions).all().get(60, TimeUnit.SECONDS)
case _ => throw new IllegalArgumentException(s"Unsupported entity type: $entityTypeHead")
}
@@ -518,7 +534,7 @@ object ConfigCommand extends Logging {
val describeAll = opts.options.has(opts.allOpt)
entityTypes.head match {
- case ConfigType.Topic | ConfigType.Broker | BrokerLoggerConfigType =>
+ case ConfigType.Topic | ConfigType.Broker | BrokerLoggerConfigType | ConfigType.ClientMetrics =>
describeResourceConfig(adminClient, entityTypes.head, entityNames.headOption, describeAll)
case ConfigType.User | ConfigType.Client =>
describeClientQuotaAndUserScramCredentialConfigs(adminClient, entityTypes, entityNames)
@@ -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")
case entityType => throw new IllegalArgumentException(s"Invalid entity type: $entityType")
})
@@ -576,6 +594,8 @@ object ConfigCommand extends Logging {
if (entityName.nonEmpty)
validateBrokerId()
(ConfigResource.Type.BROKER_LOGGER, None)
+ case ConfigType.ClientMetrics =>
+ (ConfigResource.Type.CLIENT_METRICS, Some(ConfigEntry.ConfigSource.DYNAMIC_CLIENT_METRICS_CONFIG))
case entityType => throw new IllegalArgumentException(s"Invalid entity type: $entityType")
}
@@ -791,10 +811,10 @@ object ConfigCommand extends Logging {
val describeOpt = parser.accepts("describe", "List configs for the given entity.")
val allOpt = parser.accepts("all", "List all configs for the given topic, broker, or broker-logger entity (includes static configuration when the entity type is brokers)")
- val entityType = parser.accepts("entity-type", "Type of entity (topics/clients/users/brokers/broker-loggers/ips)")
+ val entityType = parser.accepts("entity-type", "Type of entity (topics/clients/users/brokers/broker-loggers/ips/client-metrics)")
.withRequiredArg
.ofType(classOf[String])
- val entityName = parser.accepts("entity-name", "Name of entity (topic name/client id/user principal name/broker id/ip)")
+ val entityName = parser.accepts("entity-name", "Name of entity (topic name/client id/user principal name/broker id/ip/client metrics)")
.withRequiredArg
.ofType(classOf[String])
val entityDefault = parser.accepts("entity-default", "Default entity name for clients/users/brokers/ips (applies to corresponding entity type in command line)")
@@ -806,6 +826,7 @@ object ConfigCommand extends Logging {
"For entity-type '" + ConfigType.User + "': " + DynamicConfig.User.names.asScala.toSeq.sorted.map("\t" + _).mkString(nl, nl, nl) +
"For entity-type '" + ConfigType.Client + "': " + DynamicConfig.Client.names.asScala.toSeq.sorted.map("\t" + _).mkString(nl, nl, nl) +
"For entity-type '" + ConfigType.Ip + "': " + DynamicConfig.Ip.names.asScala.toSeq.sorted.map("\t" + _).mkString(nl, nl, nl) +
+ "For entity-type '" + ConfigType.ClientMetrics + "': " + DynamicConfig.ClientMetrics.names.asScala.toSeq.sorted.map("\t" + _).mkString(nl, nl, nl) +
s"Entity types '${ConfigType.User}' and '${ConfigType.Client}' may be specified together to update config for clients of a specific user.")
.withRequiredArg
.ofType(classOf[String])
@@ -939,7 +960,8 @@ object ConfigCommand extends Logging {
}
}
- if (options.has(describeOpt) && entityTypeVals.contains(BrokerLoggerConfigType) && !hasEntityName)
+ if (options.has(describeOpt) && (entityTypeVals.contains(BrokerLoggerConfigType) ||
+ entityTypeVals.contains(ConfigType.ClientMetrics)) && !hasEntityName)
throw new IllegalArgumentException(s"an entity name must be specified with --describe of ${entityTypeVals.mkString(",")}")
if (options.has(alterOpt)) {
diff --git a/core/src/main/scala/kafka/server/DynamicConfig.scala b/core/src/main/scala/kafka/server/DynamicConfig.scala
index 8af2dece042..d91a06d8bd3 100644
--- a/core/src/main/scala/kafka/server/DynamicConfig.scala
+++ b/core/src/main/scala/kafka/server/DynamicConfig.scala
@@ -25,6 +25,7 @@ import org.apache.kafka.common.config.ConfigDef.Range._
import org.apache.kafka.common.config.ConfigDef.Type._
import org.apache.kafka.storage.internals.log.LogConfig
+import java.util
import scala.jdk.CollectionConverters._
/**
@@ -111,6 +112,12 @@ object DynamicConfig {
}
}
+ object ClientMetrics {
+ private val clientConfigs = org.apache.kafka.server.metrics.ClientMetricsConfigs.configDef()
+
+ def names: util.Set[String] = clientConfigs.names
+ }
+
private def validate(configDef: ConfigDef, props: Properties, customPropsAllowed: Boolean) = {
// Validate Names
val names = configDef.names()
diff --git a/core/src/main/scala/kafka/server/ZkConfigManager.scala b/core/src/main/scala/kafka/server/ZkConfigManager.scala
index c7820859807..abead88b07c 100644
--- a/core/src/main/scala/kafka/server/ZkConfigManager.scala
+++ b/core/src/main/scala/kafka/server/ZkConfigManager.scala
@@ -39,7 +39,8 @@ object ConfigType {
val Broker = "brokers"
val Ip = "ips"
val ClientMetrics = "client-metrics"
- val all = Seq(Topic, Client, User, Broker, Ip, ClientMetrics)
+ // Do not include ClientMetrics in `all` as ClientMetrics is not supported on ZK.
+ val all = Seq(Topic, Client, User, Broker, Ip)
}
object ConfigEntityName {
diff --git a/core/src/main/scala/kafka/server/metadata/ZkConfigRepository.scala b/core/src/main/scala/kafka/server/metadata/ZkConfigRepository.scala
index 16842bcd11f..a2aabe656c5 100644
--- a/core/src/main/scala/kafka/server/metadata/ZkConfigRepository.scala
+++ b/core/src/main/scala/kafka/server/metadata/ZkConfigRepository.scala
@@ -23,6 +23,7 @@ import kafka.server.{ConfigEntityName, ConfigType}
import kafka.zk.{AdminZkClient, KafkaZkClient}
import org.apache.kafka.common.config.ConfigResource
import org.apache.kafka.common.config.ConfigResource.Type
+import org.apache.kafka.common.errors.InvalidRequestException
object ZkConfigRepository {
@@ -35,6 +36,7 @@ class ZkConfigRepository(adminZkClient: AdminZkClient) extends ConfigRepository
val configTypeForZk = configResource.`type` match {
case Type.TOPIC => ConfigType.Topic
case Type.BROKER => ConfigType.Broker
+ case Type.CLIENT_METRICS => throw new InvalidRequestException("Config type client-metrics is only supported on KRaft clusters")
case tpe => throw new IllegalArgumentException(s"Unsupported config type: $tpe")
}
// ZK stores cluster configs under "<default>".
diff --git a/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala b/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala
index 12e460a0bd3..069b79a5e0d 100644
--- a/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala
+++ b/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala
@@ -1628,6 +1628,123 @@ 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"))
+ }
+
+ private 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, util.Collections.emptyList[ConfigEntry.ConfigSynonym],
+ 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()
+ verify(alterResult).all()
+ }
+
+ @Test
+ def shouldNotDescribeClientMetricsConfigWithoutEntityName(): Unit = {
+ val describeOpts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092",
+ "--entity-type", "client-metrics",
+ "--describe"))
+
+ val exception = assertThrows(classOf[IllegalArgumentException], () => describeOpts.checkArgs())
+ assertEquals("an entity name must be specified with --describe of client-metrics", exception.getMessage)
+ }
+
+ @Test
+ def shouldNotAlterClientMetricsConfigWithoutEntityName(): Unit = {
+ val alterOpts = new ConfigCommandOptions(Array("--bootstrap-server", "localhost:9092",
+ "--entity-type", "client-metrics",
+ "--alter",
+ "--add-config", "interval.ms=1000"))
+
+ val exception = assertThrows(classOf[IllegalArgumentException], () => alterOpts.checkArgs())
+ assertEquals("an entity name must be specified with --alter of client-metrics", exception.getMessage)
+ }
+
+ @Test
+ def shouldNotSupportAlterClientMetricsWithZookeeperArg(): Unit = {
+ val alterOpts = new ConfigCommandOptions(Array("--zookeeper", zkConnect,
+ "--entity-name", "sub",
+ "--entity-type", "client-metrics",
+ "--alter",
+ "--add-config", "interval.ms=1000"))
+
+ val exception = assertThrows(classOf[IllegalArgumentException], () => alterOpts.checkArgs())
+ assertEquals("Invalid entity type client-metrics, the entity type must be one of users, brokers with a --zookeeper argument", exception.getMessage)
+ }
+
+ @Test
+ def shouldNotSupportDescribeClientMetricsWithZookeeperArg(): Unit = {
+ val describeOpts = new ConfigCommandOptions(Array("--zookeeper", zkConnect,
+ "--entity-name", "sub",
+ "--entity-type", "client-metrics",
+ "--describe"))
+
+ val exception = assertThrows(classOf[IllegalArgumentException], () => describeOpts.checkArgs())
+ assertEquals("Invalid entity type client-metrics, the entity type must be one of users, brokers with a --zookeeper argument", exception.getMessage)
+ }
+
+ @Test
+ def shouldNotSupportAlterClientMetricsWithZookeeper(): Unit = {
+ val alterOpts = new ConfigCommandOptions(Array("--zookeeper", zkConnect,
+ "--entity-name", "sub",
+ "--entity-type", "client-metrics",
+ "--alter",
+ "--add-config", "interval.ms=1000"))
+
+ val exception = assertThrows(classOf[IllegalArgumentException], () => ConfigCommand.alterConfigWithZk(null, alterOpts, dummyAdminZkClient))
+ assertEquals("client-metrics is not a known entityType. Should be one of List(topics, clients, users, brokers, ips)", exception.getMessage)
+ }
+
class DummyAdminZkClient(zkClient: KafkaZkClient) extends AdminZkClient(zkClient) {
override def changeBrokerConfig(brokerIds: Seq[Int], configs: Properties): Unit = {}
override def fetchEntityConfig(entityType: String, entityName: String): Properties = {new Properties}
diff --git a/core/src/test/scala/unit/kafka/server/metadata/ZkConfigRepositoryTest.scala b/core/src/test/scala/unit/kafka/server/metadata/ZkConfigRepositoryTest.scala
index f8737751fa5..fbb2c9e8cf9 100644
--- a/core/src/test/scala/unit/kafka/server/metadata/ZkConfigRepositoryTest.scala
+++ b/core/src/test/scala/unit/kafka/server/metadata/ZkConfigRepositoryTest.scala
@@ -22,6 +22,7 @@ import kafka.server.metadata.ZkConfigRepository
import kafka.zk.KafkaZkClient
import org.apache.kafka.common.config.ConfigResource
import org.apache.kafka.common.config.ConfigResource.Type
+import org.apache.kafka.common.errors.InvalidRequestException
import org.junit.jupiter.api.Assertions.{assertEquals, assertThrows}
import org.junit.jupiter.api.Test
import org.mockito.Mockito.{mock, when}
@@ -48,7 +49,9 @@ class ZkConfigRepositoryTest {
def testUnsupportedTypes(): Unit = {
val zkClient: KafkaZkClient = mock(classOf[KafkaZkClient])
val zkConfigRepository = ZkConfigRepository(zkClient)
- Type.values().foreach(value => if (value != Type.BROKER && value != Type.TOPIC)
+ Type.values().foreach(value => if (value != Type.BROKER && value != Type.TOPIC && value != Type.CLIENT_METRICS)
assertThrows(classOf[IllegalArgumentException], () => zkConfigRepository.config(new ConfigResource(value, value.toString))))
+ // Validate exception for CLIENT_METRICS.
+ assertThrows(classOf[InvalidRequestException], () => zkConfigRepository.config(new ConfigResource(Type.CLIENT_METRICS, Type.CLIENT_METRICS.toString)))
}
}