DL1231 commented on code in PR #19808:
URL: https://github.com/apache/kafka/pull/19808#discussion_r2106900047
##########
clients/src/test/java/org/apache/kafka/clients/admin/AdminClientTestUtils.java:
##########
@@ -163,6 +164,15 @@ public static ListConsumerGroupOffsetsResult
listConsumerGroupOffsetsResult(Stri
return new
ListConsumerGroupOffsetsResult(Collections.singletonMap(CoordinatorKey.byGroupId(group),
future));
}
+ public static ListConfigResourcesResult
listConfigResourcesResult(Map<ConfigResource.Type, Set<String>> resourceNames) {
+ Collection<ConfigResource> resources =
resourceNames.entrySet().stream()
+ .flatMap(entry -> entry.getValue().stream()
+ .map(name -> new ConfigResource(entry.getKey(), name)))
+ .collect(Collectors.toList());
+ return new
ListConfigResourcesResult(KafkaFuture.completedFuture(resources));
+
Review Comment:
nit: extra blank
##########
core/src/main/scala/kafka/admin/ConfigCommand.scala:
##########
@@ -342,6 +342,42 @@ object ConfigCommand extends Logging {
}
private def describeResourceConfig(adminClient: Admin, entityType: String,
entityName: Option[String], describeAll: Boolean): Unit = {
+ if (!describeAll) {
+ entityName.foreach { name =>
+ entityType match {
+ case TopicType =>
+ Topic.validate(name)
+ if (!adminClient.listTopics(new
ListTopicsOptions().listInternal(true)).names.get.contains(name)) {
+ System.out.println(s"The $entityType '$name' doesn't exist and
doesn't have dynamic config.")
+ return
+ }
+ case BrokerType | BrokerLoggerConfigType =>
+ if
(adminClient.describeCluster.nodes.get.stream.anyMatch(_.idString == name)) {
+ // valid broker id
+ } else if (name == BrokerDefaultEntityName) {
+ // default broker configs
+ } else {
+ System.out.println(s"The $entityType '$name' doesn't exist and
doesn't have dynamic config.")
+ return
+ }
+ case ClientMetricsType =>
+ if
(adminClient.listConfigResources(java.util.Set.of(ConfigResource.Type.CLIENT_METRICS),
new ListConfigResourcesOptions).all.get
Review Comment:
nit: Can't this just be util.Set.of?
##########
core/src/main/scala/kafka/admin/ConfigCommand.scala:
##########
@@ -342,6 +342,42 @@ object ConfigCommand extends Logging {
}
private def describeResourceConfig(adminClient: Admin, entityType: String,
entityName: Option[String], describeAll: Boolean): Unit = {
+ if (!describeAll) {
+ entityName.foreach { name =>
+ entityType match {
+ case TopicType =>
+ Topic.validate(name)
+ if (!adminClient.listTopics(new
ListTopicsOptions().listInternal(true)).names.get.contains(name)) {
+ System.out.println(s"The $entityType '$name' doesn't exist and
doesn't have dynamic config.")
+ return
+ }
+ case BrokerType | BrokerLoggerConfigType =>
+ if
(adminClient.describeCluster.nodes.get.stream.anyMatch(_.idString == name)) {
+ // valid broker id
+ } else if (name == BrokerDefaultEntityName) {
+ // default broker configs
+ } else {
+ System.out.println(s"The $entityType '$name' doesn't exist and
doesn't have dynamic config.")
+ return
+ }
+ case ClientMetricsType =>
+ if
(adminClient.listConfigResources(java.util.Set.of(ConfigResource.Type.CLIENT_METRICS),
new ListConfigResourcesOptions).all.get
+ .stream.noneMatch(_.name == name)) {
+ System.out.println(s"The $entityType '$name' doesn't exist and
doesn't have dynamic config.")
+ return
+ }
+ case GroupType =>
+ if (adminClient.listGroups().all.get.stream.noneMatch(_.groupId()
== name) &&
+
adminClient.listConfigResources(java.util.Set.of(ConfigResource.Type.GROUP),
new ListConfigResourcesOptions).all.get
Review Comment:
ditto
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
