You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "DL1231 (via GitHub)" <gi...@apache.org> on 2024/08/15 09:02:44 UTC
[PR] KAFKA-17327: Add support of group in kafka-configs.sh [kafka]
DL1231 opened a new pull request, #16887:
URL: https://github.com/apache/kafka/pull/16887
The PR adds support of alter/describe configs for group in kafka-configs.sh.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17327: Add support of group in kafka-configs.sh [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #16887:
URL: https://github.com/apache/kafka/pull/16887#discussion_r1730207204
##########
clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClient.java:
##########
@@ -243,6 +245,8 @@ private MockAdminClient(
this.brokerLogDirs = brokerLogDirs;
this.brokerConfigs = new ArrayList<>();
this.clientMetricsConfigs = new HashMap<>();
+ this.groupConfigs = new HashMap<>();
+ this.defaultGroupConfigs = generateDefaultGroupConfigs();
Review Comment:
Will do, thanks for the suggestion.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17327: Add support of group in kafka-configs.sh [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on PR #16887:
URL: https://github.com/apache/kafka/pull/16887#issuecomment-2311385727
@dajac @chia7712 , please take a look when available. Thank you.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17327: Add support of group in kafka-configs.sh [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #16887:
URL: https://github.com/apache/kafka/pull/16887#discussion_r1719887934
##########
core/src/main/scala/kafka/admin/ConfigCommand.scala:
##########
@@ -562,6 +543,8 @@ object ConfigCommand extends Logging {
adminClient.describeCluster(new DescribeClusterOptions()).nodes().get().asScala.map(_.idString).toSeq :+ BrokerDefaultEntityName
case ConfigType.CLIENT_METRICS =>
adminClient.listClientMetricsResources().all().get().asScala.map(_.name).toSeq
+ case ConfigType.GROUP =>
+ adminClient.listConsumerGroups(new ListConsumerGroupsOptions().withTypes(Seq(GroupType.CONSUMER).toSet.asJava)).all.get.asScala.map(_.groupId).toSeq
Review Comment:
@chia7712 Group configs can be applied to all groups. KIP-932 introduces 5 new group configs which I expect to deliver in the coming days. For example, isolation level for a share group is a group-level config configured with `group.share.isolation.level`.
KIP-1043 would enable you to list groups of all types in a cleaner way because it introduces `AdminClient.listGroups()`. The thing I struggle with here is that `adminClient.listConsumerGroups()` sounds like it actually returns groups of all kinds, but that's misleading. The result might contain groups which are not consumer groups and if you attempt consumer group operations on them, you'll get errors. It should return consumer groups.
Obviously, there's some history here and the transition as the new consumer group protocol and the new group types are introduced has a few rough edges because these were not ideas that were anticipated a few years ago.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17327: Add support of group in kafka-configs.sh [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on PR #16887:
URL: https://github.com/apache/kafka/pull/16887#issuecomment-2293496752
@DL1231 Thanks for the patch. Would it be possible to add unit and integration tests?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17327: Add support of group in kafka-configs.sh [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #16887:
URL: https://github.com/apache/kafka/pull/16887#discussion_r1724977996
##########
core/src/main/scala/kafka/admin/ConfigCommand.scala:
##########
@@ -868,6 +871,9 @@ object ConfigCommand extends Logging {
val ip: OptionSpec[String] = parser.accepts("ip", "The IP address.")
.withRequiredArg
.ofType(classOf[String])
+ val group: OptionSpec[String] = parser.accepts("group", "The group's name.")
Review Comment:
This is a group ID really, not a group name.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17327: Add support of group in kafka-configs.sh [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #16887:
URL: https://github.com/apache/kafka/pull/16887#discussion_r1718343472
##########
core/src/main/scala/kafka/admin/ConfigCommand.scala:
##########
@@ -562,6 +543,8 @@ object ConfigCommand extends Logging {
adminClient.describeCluster(new DescribeClusterOptions()).nodes().get().asScala.map(_.idString).toSeq :+ BrokerDefaultEntityName
case ConfigType.CLIENT_METRICS =>
adminClient.listClientMetricsResources().all().get().asScala.map(_.name).toSeq
+ case ConfigType.GROUP =>
+ adminClient.listConsumerGroups(new ListConsumerGroupsOptions().withTypes(Seq(GroupType.CONSUMER).toSet.asJava)).all.get.asScala.map(_.groupId).toSeq
Review Comment:
This is interesting. Because you've used `withTypes(GroupType.CONSUMER)`, this only lists out the consumers which are using the new consumer group protocol. So, I can set configs for classic consumer groups, but `--describe` without a group name will not show them.
Really, the solution here I think is in KIP-1043 which adds `adminClient.listGroups()` so that all groups regardless of type can be listed. Once that KIP is approved, I'll improve this area. For now, I think `Seq(GroupType.CONSUMER, GroupType.CLASSIC)` is probably the best you can do. That does at least list out the configs for all new and classic consumer groups.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17327: Add support of group in kafka-configs.sh [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #16887:
URL: https://github.com/apache/kafka/pull/16887#discussion_r1730575833
##########
core/src/test/java/kafka/admin/ConfigCommandIntegrationTest.java:
##########
@@ -256,6 +278,32 @@ public void testDynamicBrokerConfigUpdateUsingKraft() throws Exception {
}
}
+ @ClusterTest(types = {Type.KRAFT, Type.CO_KRAFT}, serverProperties = {
+ @ClusterConfigProperty(key = "group.coordinator.new.enable", value = "true"),
+ @ClusterConfigProperty(key = "group.coordinator.rebalance.protocols", value = "classic,consumer"),
Review Comment:
Done.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17327: Add support of group in kafka-configs.sh [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on PR #16887:
URL: https://github.com/apache/kafka/pull/16887#issuecomment-2301753012
Hi @dajac @chia7712 @AndrewJSchofield . Thanks a lot for the review. I have made the required changes in the last commit, PTAL when you get a chance.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17327: Add support of group in kafka-configs.sh [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16887:
URL: https://github.com/apache/kafka/pull/16887#discussion_r1721014948
##########
core/src/main/scala/kafka/admin/ConfigCommand.scala:
##########
@@ -580,6 +563,23 @@ object ConfigCommand extends Logging {
}
}
+ private def alterResourceConfig(adminClient: Admin, entityTypeHead: String, entityNameHead: String, configsToBeDeleted: Seq[String], configsToBeAdded: Map[String, ConfigEntry], resourceType: ConfigResource.Type): Unit = {
+ val oldConfig = getResourceConfig(adminClient, entityTypeHead, entityNameHead, includeSynonyms = false, describeAll = false)
Review Comment:
> This is not wrong because the group could be create any time and it is, it would have those configurations by default.
My concern was the consistency. The type `TOPIC` has similar attributes but we return exception if it is nonexistent [0]. However, maybe we can add enough docs to the APIs to remind users
[0] https://github.com/apache/kafka/blob/c5e91546721b6b160732667ca7450d22927cc3b3/core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala#L1513
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17327: Add support of group in kafka-configs.sh [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16887:
URL: https://github.com/apache/kafka/pull/16887#discussion_r1721018890
##########
core/src/main/scala/kafka/admin/ConfigCommand.scala:
##########
@@ -580,6 +563,23 @@ object ConfigCommand extends Logging {
}
}
+ private def alterResourceConfig(adminClient: Admin, entityTypeHead: String, entityNameHead: String, configsToBeDeleted: Seq[String], configsToBeAdded: Map[String, ConfigEntry], resourceType: ConfigResource.Type): Unit = {
+ val oldConfig = getResourceConfig(adminClient, entityTypeHead, entityNameHead, includeSynonyms = false, describeAll = false)
Review Comment:
https://issues.apache.org/jira/browse/KAFKA-17359 is opened
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17327: Add support of group in kafka-configs.sh [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on PR #16887:
URL: https://github.com/apache/kafka/pull/16887#issuecomment-2301893041
> This PR is looking quite good now but I notice that there's one behaviour which doesn't seem to be supported.
>
> If I use `bin/kafka-configs.sh --bootstrap-server localhost:9092 --entity-type groups --describe`, I think I would expect it to list the group resources for which there are configs defined and their configs. What it does is list the groups which exist and their configs, but it omits group resources for which configs are defined but there is no actual group in existence at this time.
>
> It might take a KIP to fix this (which of course would be beyond the scope of this PR), but this does seem like the right place to begin the discussion at least. For topics, you can only define configs if the topic exists. For client metrics, the resource comes into existence as a result of defining the configs. For groups, there are essentially two resources - a config resource and the group itself.
Perhaps we could introduce a new RPC to the Kafka protocol called 'ListGroupConfigResources', which responds with a list of group configuration resources, regardless of the group's existence.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17327: Add support of group in kafka-configs.sh [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #16887:
URL: https://github.com/apache/kafka/pull/16887#discussion_r1719897297
##########
core/src/main/scala/kafka/admin/ConfigCommand.scala:
##########
@@ -580,6 +563,23 @@ object ConfigCommand extends Logging {
}
}
+ private def alterResourceConfig(adminClient: Admin, entityTypeHead: String, entityNameHead: String, configsToBeDeleted: Seq[String], configsToBeAdded: Map[String, ConfigEntry], resourceType: ConfigResource.Type): Unit = {
+ val oldConfig = getResourceConfig(adminClient, entityTypeHead, entityNameHead, includeSynonyms = false, describeAll = false)
Review Comment:
That's probably the best behaviour.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17327: Add support of group in kafka-configs.sh [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #16887:
URL: https://github.com/apache/kafka/pull/16887#discussion_r1719829270
##########
core/src/main/scala/kafka/admin/ConfigCommand.scala:
##########
@@ -562,6 +543,8 @@ object ConfigCommand extends Logging {
adminClient.describeCluster(new DescribeClusterOptions()).nodes().get().asScala.map(_.idString).toSeq :+ BrokerDefaultEntityName
case ConfigType.CLIENT_METRICS =>
adminClient.listClientMetricsResources().all().get().asScala.map(_.name).toSeq
+ case ConfigType.GROUP =>
+ adminClient.listConsumerGroups(new ListConsumerGroupsOptions().withTypes(Seq(GroupType.CONSUMER).toSet.asJava)).all.get.asScala.map(_.groupId).toSeq
Review Comment:
I would not specify any types here as the dynamic configs should work with all the existing group types and all the future group types too. I agree that the name of the method is unfortunate here because the method lists all groups.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17327: Add support of group in kafka-configs.sh [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16887:
URL: https://github.com/apache/kafka/pull/16887#discussion_r1723342987
##########
core/src/test/java/kafka/admin/ConfigCommandIntegrationTest.java:
##########
@@ -544,6 +615,18 @@ private void verifyPerBrokerConfigValue(Admin client,
}, 5000, config + " are not updated");
}
+ private void verifyDeletedGroupConfig(Admin client,
Review Comment:
According to https://github.com/apache/kafka/pull/16887#discussion_r1719827276, we are always able to see the group configs. Hence, we should check that the value is revert to "default value" after we delete the configs.
##########
clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClient.java:
##########
@@ -824,6 +828,13 @@ private synchronized Config getResourceDescription(ConfigResource resource) {
}
return toConfigObject(clientMetricsConfigs.get(resourceName));
}
+ case GROUP: {
+ String resourceName = resource.name();
+ if (resourceName.isEmpty()) {
+ throw new InvalidRequestException("Empty resource name");
+ }
+ return toConfigObject(groupConfigs.get(resourceName));
Review Comment:
Should we align the behavior to return "default configs" even though the group is nonexistent?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17327: Add support of group in kafka-configs.sh [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #16887:
URL: https://github.com/apache/kafka/pull/16887#discussion_r1729122211
##########
core/src/test/java/kafka/admin/ConfigCommandIntegrationTest.java:
##########
@@ -256,6 +278,32 @@ public void testDynamicBrokerConfigUpdateUsingKraft() throws Exception {
}
}
+ @ClusterTest(types = {Type.KRAFT, Type.CO_KRAFT}, serverProperties = {
+ @ClusterConfigProperty(key = "group.coordinator.new.enable", value = "true"),
+ @ClusterConfigProperty(key = "group.coordinator.rebalance.protocols", value = "classic,consumer"),
Review Comment:
Can we use the constants instead of the raw strings? It should work from here.
##########
clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClient.java:
##########
@@ -243,6 +245,8 @@ private MockAdminClient(
this.brokerLogDirs = brokerLogDirs;
this.brokerConfigs = new ArrayList<>();
this.clientMetricsConfigs = new HashMap<>();
+ this.groupConfigs = new HashMap<>();
+ this.defaultGroupConfigs = generateDefaultGroupConfigs();
Review Comment:
Should we rather define a constant for the default group configs? Thinking a bit more about those too. I wonder if we should rather extend the builder to let the user passes the default group config that he wants to use instead of having some defaults defined here. Then, we can have a constant on the user side which is a litter clearer in my opinion because you see them when you read the tests. What do you think?
##########
core/src/test/java/kafka/admin/ConfigCommandIntegrationTest.java:
##########
@@ -126,6 +137,17 @@ public void testNullStatusOnKraftCommandAlterUserQuota() {
assertTrue(StringUtils.isBlank(message), message);
}
+ @ClusterTest(types = {Type.CO_KRAFT, Type.KRAFT})
+ public void testNullStatusOnKraftCommandAlterGroup() {
+ Stream<String> command = Stream.concat(quorumArgs(), Stream.of(
+ "--entity-type", "groups",
+ "--entity-name", "group",
Review Comment:
Should we also test with the `--group` alias?
##########
core/src/test/java/kafka/admin/ConfigCommandTest.java:
##########
@@ -1918,6 +1943,148 @@ public void shouldNotSupportAlterClientMetricsWithZookeeper() {
assertEquals("client-metrics is not a known entityType. Should be one of List(topics, clients, users, brokers, ips)", exception.getMessage());
}
+ @Test
+ public void shouldAlterGroupConfig() {
+ Node node = new Node(1, "localhost", 9092);
+ verifyAlterGroupConfig(node, "group", Arrays.asList("--entity-name", "group"));
+ }
+
+ private void verifyAlterGroupConfig(Node node, String resourceName, List<String> resourceOpts) {
+ List<String> optsList = concat(Arrays.asList("--bootstrap-server", "localhost:9092",
+ "--entity-type", "groups",
+ "--alter",
+ "--delete-config", "consumer.session.timeout.ms",
+ "--add-config", "consumer.heartbeat.interval.ms=6000"), resourceOpts);
+ ConfigCommand.ConfigCommandOptions alterOpts = new ConfigCommand.ConfigCommandOptions(toArray(optsList));
+
+ ConfigResource resource = new ConfigResource(ConfigResource.Type.GROUP, resourceName);
+ List<ConfigEntry> configEntries = Collections.singletonList(new ConfigEntry("consumer.session.timeout.ms", "45000",
+ ConfigEntry.ConfigSource.DYNAMIC_GROUP_CONFIG, false, false, Collections.emptyList(),
+ ConfigEntry.ConfigType.UNKNOWN, null));
+ KafkaFutureImpl<Map<ConfigResource, Config>> future = new KafkaFutureImpl<>();
+ future.complete(Collections.singletonMap(resource, new Config(configEntries)));
+ DescribeConfigsResult describeResult = mock(DescribeConfigsResult.class);
+ when(describeResult.all()).thenReturn(future);
+
+ KafkaFutureImpl<Void> alterFuture = new KafkaFutureImpl<>();
+ alterFuture.complete(null);
+ AlterConfigsResult alterResult = mock(AlterConfigsResult.class);
+ when(alterResult.all()).thenReturn(alterFuture);
+
+ MockAdminClient mockAdminClient = new MockAdminClient(Collections.singletonList(node), node) {
+ @Override
+ public synchronized DescribeConfigsResult describeConfigs(Collection<ConfigResource> resources, DescribeConfigsOptions options) {
+ assertFalse(options.includeSynonyms(), "Config synonyms requested unnecessarily");
+ assertEquals(1, resources.size());
+ ConfigResource res = resources.iterator().next();
+ assertEquals(ConfigResource.Type.GROUP, res.type());
+ assertEquals(resourceName, res.name());
+ return describeResult;
+ }
+
+ @Override
+ public synchronized AlterConfigsResult incrementalAlterConfigs(Map<ConfigResource, Collection<AlterConfigOp>> configs, AlterConfigsOptions options) {
+ assertEquals(1, configs.size());
+ Map.Entry<ConfigResource, Collection<AlterConfigOp>> entry = configs.entrySet().iterator().next();
+ ConfigResource res = entry.getKey();
+ Collection<AlterConfigOp> alterConfigOps = entry.getValue();
+ assertEquals(ConfigResource.Type.GROUP, res.type());
+ assertEquals(2, alterConfigOps.size());
+
+ List<AlterConfigOp> expectedConfigOps = Arrays.asList(
+ new AlterConfigOp(new ConfigEntry("consumer.heartbeat.interval.ms", "6000"), AlterConfigOp.OpType.SET),
+ new AlterConfigOp(new ConfigEntry("consumer.session.timeout.ms", ""), AlterConfigOp.OpType.DELETE)
+ );
+ assertEquals(expectedConfigOps.size(), alterConfigOps.size());
+ Iterator<AlterConfigOp> alterConfigOpsIter = alterConfigOps.iterator();
+ for (AlterConfigOp expectedConfigOp : expectedConfigOps) {
+ assertEquals(expectedConfigOp, alterConfigOpsIter.next());
+ }
+ return alterResult;
+ }
+ };
+ ConfigCommand.alterConfig(mockAdminClient, alterOpts);
+ verify(describeResult).all();
+ verify(alterResult).all();
+ }
+
+ @Test
+ public void shouldDescribeGroupConfigWithoutEntityName() {
+ ConfigCommand.ConfigCommandOptions describeOpts = new ConfigCommand.ConfigCommandOptions(toArray("--bootstrap-server", "localhost:9092",
+ "--entity-type", "groups",
+ "--describe"));
+
+ ConfigResource resourceCustom = new ConfigResource(ConfigResource.Type.GROUP, "group");
+ ConfigEntry configEntry = new ConfigEntry("consumer.heartbeat.interval.ms", "6000");
+ KafkaFutureImpl<Map<ConfigResource, Config>> future = new KafkaFutureImpl<>();
+ DescribeConfigsResult describeResult = mock(DescribeConfigsResult.class);
+ when(describeResult.all()).thenReturn(future);
+
+ Node node = new Node(1, "localhost", 9092);
+ MockAdminClient mockAdminClient = new MockAdminClient(Collections.singletonList(node), node) {
+ @Override
+ public synchronized DescribeConfigsResult describeConfigs(Collection<ConfigResource> resources, DescribeConfigsOptions options) {
+ assertTrue(options.includeSynonyms());
+ assertEquals(1, resources.size());
+ ConfigResource resource = resources.iterator().next();
+ assertEquals(ConfigResource.Type.GROUP, resource.type());
+ assertEquals(resourceCustom.name(), resource.name());
+ future.complete(Collections.singletonMap(resourceCustom, new Config(Collections.singletonList(configEntry))));
+ return describeResult;
+ }
+ };
+ mockAdminClient.incrementalAlterConfigs(Collections.singletonMap(resourceCustom,
+ Collections.singletonList(new AlterConfigOp(configEntry, AlterConfigOp.OpType.SET))), new AlterConfigsOptions());
+ ConfigCommand.describeConfig(mockAdminClient, describeOpts);
+ verify(describeResult).all();
+ }
+
+ @Test
+ public void shouldNotAlterGroupConfigWithoutEntityName() {
+ ConfigCommand.ConfigCommandOptions alterOpts = new ConfigCommand.ConfigCommandOptions(toArray("--bootstrap-server", "localhost:9092",
+ "--entity-type", "groups",
+ "--alter",
+ "--add-config", "consumer.heartbeat.interval.ms=6000"));
+
+ IllegalArgumentException exception = assertThrows(IllegalArgumentException.class, alterOpts::checkArgs);
+ assertEquals("an entity name must be specified with --alter of groups", exception.getMessage());
+ }
+
+ @Test
+ public void shouldNotSupportAlterGroupConfigWithZookeeperArg() {
+ ConfigCommand.ConfigCommandOptions alterOpts = new ConfigCommand.ConfigCommandOptions(toArray("--zookeeper", ZK_CONNECT,
+ "--entity-name", "group",
+ "--entity-type", "groups",
+ "--alter",
+ "--add-config", "consumer.heartbeat.interval.ms=6000"));
+
+ IllegalArgumentException exception = assertThrows(IllegalArgumentException.class, alterOpts::checkArgs);
+ assertEquals("Invalid entity type groups, the entity type must be one of users, brokers with a --zookeeper argument", exception.getMessage());
+ }
+
+ @Test
+ public void shouldNotSupportDescribeGroupConfigWithZookeeperArg() {
+ ConfigCommand.ConfigCommandOptions describeOpts = new ConfigCommand.ConfigCommandOptions(toArray("--zookeeper", ZK_CONNECT,
+ "--entity-name", "group",
+ "--entity-type", "groups",
+ "--describe"));
+
+ IllegalArgumentException exception = assertThrows(IllegalArgumentException.class, describeOpts::checkArgs);
+ assertEquals("Invalid entity type groups, the entity type must be one of users, brokers with a --zookeeper argument", exception.getMessage());
+ }
+
+ @Test
+ public void shouldNotSupportAlterGroupConfigWithZookeeper() {
+ ConfigCommand.ConfigCommandOptions alterOpts = new ConfigCommand.ConfigCommandOptions(toArray("--zookeeper", ZK_CONNECT,
+ "--entity-name", "group",
+ "--entity-type", "groups",
+ "--alter",
+ "--add-config", "consumer.heartbeat.interval.ms=6000"));
+
+ IllegalArgumentException exception = assertThrows(IllegalArgumentException.class, () -> ConfigCommand.alterConfigWithZk(null, alterOpts, DUMMY_ADMIN_ZK_CLIENT));
+ assertEquals("groups is not a known entityType. Should be one of List(topics, clients, users, brokers, ips)", exception.getMessage());
+ }
+
Review Comment:
Should we also add a unit test for the `--group` alias?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17327: Add support of group in kafka-configs.sh [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16887:
URL: https://github.com/apache/kafka/pull/16887#discussion_r1730446002
##########
core/src/test/java/kafka/admin/ConfigCommandIntegrationTest.java:
##########
@@ -256,6 +278,32 @@ public void testDynamicBrokerConfigUpdateUsingKraft() throws Exception {
}
}
+ @ClusterTest(types = {Type.KRAFT, Type.CO_KRAFT}, serverProperties = {
+ @ClusterConfigProperty(key = "group.coordinator.new.enable", value = "true"),
+ @ClusterConfigProperty(key = "group.coordinator.rebalance.protocols", value = "classic,consumer"),
Review Comment:
@DL1231 I feel that is false limit. Could you adjust the checkstyle rule?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17327: Add support of group in kafka-configs.sh [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #16887:
URL: https://github.com/apache/kafka/pull/16887#discussion_r1730271755
##########
core/src/test/java/kafka/admin/ConfigCommandIntegrationTest.java:
##########
@@ -256,6 +278,32 @@ public void testDynamicBrokerConfigUpdateUsingKraft() throws Exception {
}
}
+ @ClusterTest(types = {Type.KRAFT, Type.CO_KRAFT}, serverProperties = {
+ @ClusterConfigProperty(key = "group.coordinator.new.enable", value = "true"),
+ @ClusterConfigProperty(key = "group.coordinator.rebalance.protocols", value = "classic,consumer"),
Review Comment:
I tried to do it, but failed.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17327: Add support of group in kafka-configs.sh [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16887:
URL: https://github.com/apache/kafka/pull/16887#discussion_r1718910571
##########
core/src/test/java/kafka/admin/ConfigCommandTest.java:
##########
@@ -1918,6 +1918,148 @@ public void shouldNotSupportAlterClientMetricsWithZookeeper() {
assertEquals("client-metrics is not a known entityType. Should be one of List(topics, clients, users, brokers, ips)", exception.getMessage());
}
+ @Test
+ public void shouldAlterGroupConfig() {
Review Comment:
Do you plan to add IT for support of group? or it is a future work?
##########
core/src/main/scala/kafka/admin/ConfigCommand.scala:
##########
@@ -562,6 +543,8 @@ object ConfigCommand extends Logging {
adminClient.describeCluster(new DescribeClusterOptions()).nodes().get().asScala.map(_.idString).toSeq :+ BrokerDefaultEntityName
case ConfigType.CLIENT_METRICS =>
adminClient.listClientMetricsResources().all().get().asScala.map(_.name).toSeq
+ case ConfigType.GROUP =>
+ adminClient.listConsumerGroups(new ListConsumerGroupsOptions().withTypes(Seq(GroupType.CONSUMER).toSet.asJava)).all.get.asScala.map(_.groupId).toSeq
Review Comment:
> So, I can set configs for classic consumer groups, but --describe without a group name will not show them.
does `Share` have similar issue?
> Once that KIP is approved, I'll improve this area.
our of curiosity. I feel your suggestion `Seq(GroupType.CONSUMER, GroupType.CLASSIC)` is good enough if we only care about the configs of `CONSUMER` and `CLASSIC` . what is the improvement of this case introduced by KIP-1043?
##########
core/src/main/scala/kafka/admin/ConfigCommand.scala:
##########
@@ -580,6 +563,23 @@ object ConfigCommand extends Logging {
}
}
+ private def alterResourceConfig(adminClient: Admin, entityTypeHead: String, entityNameHead: String, configsToBeDeleted: Seq[String], configsToBeAdded: Map[String, ConfigEntry], resourceType: ConfigResource.Type): Unit = {
+ val oldConfig = getResourceConfig(adminClient, entityTypeHead, entityNameHead, includeSynonyms = false, describeAll = false)
Review Comment:
In reviewing it, I notice that `admin.describeConfigs` will return static configs even if the group is nonexistent. IMHO, that should return GROUP_ID_NOT_FOUND. @AndrewJSchofield @DL1231 WDYT? I can file a jira to fix it if it is NOT expected behavior.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17327: Add support of group in kafka-configs.sh [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac merged PR #16887:
URL: https://github.com/apache/kafka/pull/16887
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17327: Add support of group in kafka-configs.sh [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on PR #16887:
URL: https://github.com/apache/kafka/pull/16887#issuecomment-2295504958
@AndrewJSchofield @chia7712 @halorgium Thanks for your comments and please excuse me for the delay on this one. I will address your comments asap.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17327: Add support of group in kafka-configs.sh [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #16887:
URL: https://github.com/apache/kafka/pull/16887#discussion_r1718343472
##########
core/src/main/scala/kafka/admin/ConfigCommand.scala:
##########
@@ -562,6 +543,8 @@ object ConfigCommand extends Logging {
adminClient.describeCluster(new DescribeClusterOptions()).nodes().get().asScala.map(_.idString).toSeq :+ BrokerDefaultEntityName
case ConfigType.CLIENT_METRICS =>
adminClient.listClientMetricsResources().all().get().asScala.map(_.name).toSeq
+ case ConfigType.GROUP =>
+ adminClient.listConsumerGroups(new ListConsumerGroupsOptions().withTypes(Seq(GroupType.CONSUMER).toSet.asJava)).all.get.asScala.map(_.groupId).toSeq
Review Comment:
This is interesting. Because you've used `withTypes(GroupType.CONSUMER)`, this only lists out the consumers which are using the new consumer group protocol. So, I can set configs for other consumers, but `--describe` without a group name will not show them.
Really, the solution here I think is in KIP-1043 which adds `adminClient.listGroups()` so that all groups regardless of type can be listed. Once that KIP is approved, I'll improve this area. For now, I think `Seq(GroupType.CONSUMER, GroupType.CLASSIC)` is probably the best you can do. That does at least list out the configs for all new and classic consumer groups.
##########
core/src/main/scala/kafka/admin/ConfigCommand.scala:
##########
@@ -57,6 +58,7 @@ import scala.collection._
* <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>
+ * <li> group: --group <group> OR --entiry-type groups --entity-name <group>
Review Comment:
Also, you didn't add the `--group` option. I think you should either add that option, or change this comment to remove `--group`.
I've noticed that I didn't implement `--client-metrics` when I did `--entity-type client-metrics`. I'll open a JIRA to fix that.
##########
core/src/main/scala/kafka/admin/ConfigCommand.scala:
##########
@@ -57,6 +58,7 @@ import scala.collection._
* <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>
+ * <li> group: --group <group> OR --entiry-type groups --entity-name <group>
Review Comment:
nit: "entiry" should be "entity"
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17327: Add support of group in kafka-configs.sh [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #16887:
URL: https://github.com/apache/kafka/pull/16887#discussion_r1719827276
##########
core/src/main/scala/kafka/admin/ConfigCommand.scala:
##########
@@ -580,6 +563,23 @@ object ConfigCommand extends Logging {
}
}
+ private def alterResourceConfig(adminClient: Admin, entityTypeHead: String, entityNameHead: String, configsToBeDeleted: Seq[String], configsToBeAdded: Map[String, ConfigEntry], resourceType: ConfigResource.Type): Unit = {
+ val oldConfig = getResourceConfig(adminClient, entityTypeHead, entityNameHead, includeSynonyms = false, describeAll = false)
Review Comment:
I thought about this do while review the describe PR. It is a bit tricky because the controller does not know whether the group really exists or not. Only the group coordinator does know this information. This is why I leaned towards returning the default static configs in this case. This is not wrong because the group could be create any time and it is, it would have those configurations by default.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org