You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "nizhikov (via GitHub)" <gi...@apache.org> on 2024/01/24 17:19:53 UTC
[PR] KAFKA-14589 [1/3][WIP] Tests of ConsoleGroupCommand rewritten in java [kafka]
nizhikov opened a new pull request, #15256:
URL: https://github.com/apache/kafka/pull/15256
*More detailed description of your change,
if necessary. The PR title and PR message become
the squashed commit message, so use a separate
comment to ping reviewers.*
*Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.*
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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-14589 [1/3] Tests of ConsoleGroupCommand rewritten in java [kafka]
Posted by "nizhikov (via GitHub)" <gi...@apache.org>.
nizhikov commented on PR #15256:
URL: https://github.com/apache/kafka/pull/15256#issuecomment-1921091122
> could you rebase with the latest trunk branch and let's see if it fixed the test?
PR currently merged with the latest trunk changes. https://github.com/apache/kafka/pull/15256/commits/74980be0279f908835ef2ce9f5b1c0aabdf8dcec
--
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-14589 [1/3] Tests of ConsoleGroupCommand rewritten in java [kafka]
Posted by "nizhikov (via GitHub)" <gi...@apache.org>.
nizhikov commented on code in PR #15256:
URL: https://github.com/apache/kafka/pull/15256#discussion_r1471198261
##########
tools/src/test/java/org/apache/kafka/tools/consumer/group/ListConsumerGroupTest.java:
##########
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.tools.consumer.group;
+
+import joptsimple.OptionException;
+import kafka.admin.ConsumerGroupCommand;
+import org.apache.kafka.clients.admin.ConsumerGroupListing;
+import org.apache.kafka.common.ConsumerGroupState;
+import org.apache.kafka.test.TestUtils;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Objects;
+import java.util.Optional;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class ListConsumerGroupTest extends ConsumerGroupCommandTest {
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListConsumerGroups(String quorum) throws Exception {
+ String simpleGroup = "simple-group";
+ addSimpleGroupExecutor(simpleGroup);
+ addConsumerGroupExecutor(1);
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list"};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+ scala.collection.Set<String> expectedGroups = set(Arrays.asList(GROUP, simpleGroup));
+ final scala.collection.Set[] foundGroups = new scala.collection.Set[1];
+ TestUtils.waitForCondition(() -> {
+ foundGroups[0] = service.listConsumerGroups().toSet();
+ return Objects.equals(expectedGroups, foundGroups[0]);
+ }, "Expected --list to show groups " + expectedGroups + ", but found " + foundGroups[0] + ".");
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListWithUnrecognizedNewConsumerOption() {
+ String[] cgcArgs = new String[]{"--new-consumer", "--bootstrap-server", bootstrapServers(listenerName()), "--list"};
+ assertThrows(OptionException.class, () -> getConsumerGroupService(cgcArgs));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListConsumerGroupsWithStates() throws Exception {
+ String simpleGroup = "simple-group";
+ addSimpleGroupExecutor(simpleGroup);
+ addConsumerGroupExecutor(1);
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list", "--state"};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ scala.collection.Set<ConsumerGroupListing> expectedListing = set(Arrays.asList(
+ new ConsumerGroupListing(simpleGroup, true, Optional.of(ConsumerGroupState.EMPTY)),
+ new ConsumerGroupListing(GROUP, false, Optional.of(ConsumerGroupState.STABLE))));
+
+ final scala.collection.Set[] foundListing = new scala.collection.Set[1];
+ TestUtils.waitForCondition(() -> {
+ foundListing[0] = service.listConsumerGroupsWithState(set(Arrays.asList(ConsumerGroupState.values()))).toSet();
+ return Objects.equals(expectedListing, foundListing[0]);
+ }, "Expected to show groups " + expectedListing + ", but found " + foundListing[0]);
+
+ scala.collection.Set<ConsumerGroupListing> expectedListingStable = set(Collections.singleton(
+ new ConsumerGroupListing(GROUP, false, Optional.of(ConsumerGroupState.STABLE))));
+
+ foundListing[0] = null;
+
+ TestUtils.waitForCondition(() -> {
+ foundListing[0] = service.listConsumerGroupsWithState(set(Collections.singleton(ConsumerGroupState.STABLE))).toSet();
+ return Objects.equals(expectedListingStable, foundListing[0]);
+ }, "Expected to show groups " + expectedListingStable + ", but found " + foundListing[0]);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testConsumerGroupStatesFromString(String quorum) {
+ scala.collection.Set<ConsumerGroupState> result = ConsumerGroupCommand.consumerGroupStatesFromString("Stable");
+ assertEquals(set(Collections.singleton(ConsumerGroupState.STABLE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("Stable, PreparingRebalance");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.STABLE, ConsumerGroupState.PREPARING_REBALANCE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("Dead,CompletingRebalance,");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.DEAD, ConsumerGroupState.COMPLETING_REBALANCE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("stable");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.STABLE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("stable, assigning");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.STABLE, ConsumerGroupState.ASSIGNING)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("dead,reconciling,");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.DEAD, ConsumerGroupState.RECONCILING)), result);
+
+ assertThrows(IllegalArgumentException.class, () -> ConsumerGroupCommand.consumerGroupStatesFromString("bad, wrong"));
+
+ assertThrows(IllegalArgumentException.class, () -> ConsumerGroupCommand.consumerGroupStatesFromString(" bad, Stable"));
+
+ assertThrows(IllegalArgumentException.class, () -> ConsumerGroupCommand.consumerGroupStatesFromString(" , ,"));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListGroupCommand(String quorum) throws Exception {
+ String simpleGroup = "simple-group";
+ addSimpleGroupExecutor(simpleGroup);
+ addConsumerGroupExecutor(1);
+ final String[] out = {""};
Review Comment:
out value changed in java lambda
`out = grabConsoleOutput()` and only final variable (which can't be changed) can be used inside java lambda.
I can rework String array to usage of `AtomicReference`.
What do you think?
##########
tools/src/test/java/org/apache/kafka/tools/consumer/group/ListConsumerGroupTest.java:
##########
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.tools.consumer.group;
+
+import joptsimple.OptionException;
+import kafka.admin.ConsumerGroupCommand;
+import org.apache.kafka.clients.admin.ConsumerGroupListing;
+import org.apache.kafka.common.ConsumerGroupState;
+import org.apache.kafka.test.TestUtils;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Objects;
+import java.util.Optional;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class ListConsumerGroupTest extends ConsumerGroupCommandTest {
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListConsumerGroups(String quorum) throws Exception {
+ String simpleGroup = "simple-group";
+ addSimpleGroupExecutor(simpleGroup);
+ addConsumerGroupExecutor(1);
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list"};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+ scala.collection.Set<String> expectedGroups = set(Arrays.asList(GROUP, simpleGroup));
+ final scala.collection.Set[] foundGroups = new scala.collection.Set[1];
+ TestUtils.waitForCondition(() -> {
+ foundGroups[0] = service.listConsumerGroups().toSet();
+ return Objects.equals(expectedGroups, foundGroups[0]);
+ }, "Expected --list to show groups " + expectedGroups + ", but found " + foundGroups[0] + ".");
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListWithUnrecognizedNewConsumerOption() {
+ String[] cgcArgs = new String[]{"--new-consumer", "--bootstrap-server", bootstrapServers(listenerName()), "--list"};
+ assertThrows(OptionException.class, () -> getConsumerGroupService(cgcArgs));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListConsumerGroupsWithStates() throws Exception {
+ String simpleGroup = "simple-group";
+ addSimpleGroupExecutor(simpleGroup);
+ addConsumerGroupExecutor(1);
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list", "--state"};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ scala.collection.Set<ConsumerGroupListing> expectedListing = set(Arrays.asList(
+ new ConsumerGroupListing(simpleGroup, true, Optional.of(ConsumerGroupState.EMPTY)),
+ new ConsumerGroupListing(GROUP, false, Optional.of(ConsumerGroupState.STABLE))));
+
+ final scala.collection.Set[] foundListing = new scala.collection.Set[1];
+ TestUtils.waitForCondition(() -> {
+ foundListing[0] = service.listConsumerGroupsWithState(set(Arrays.asList(ConsumerGroupState.values()))).toSet();
+ return Objects.equals(expectedListing, foundListing[0]);
+ }, "Expected to show groups " + expectedListing + ", but found " + foundListing[0]);
+
+ scala.collection.Set<ConsumerGroupListing> expectedListingStable = set(Collections.singleton(
+ new ConsumerGroupListing(GROUP, false, Optional.of(ConsumerGroupState.STABLE))));
+
+ foundListing[0] = null;
+
+ TestUtils.waitForCondition(() -> {
+ foundListing[0] = service.listConsumerGroupsWithState(set(Collections.singleton(ConsumerGroupState.STABLE))).toSet();
+ return Objects.equals(expectedListingStable, foundListing[0]);
+ }, "Expected to show groups " + expectedListingStable + ", but found " + foundListing[0]);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testConsumerGroupStatesFromString(String quorum) {
+ scala.collection.Set<ConsumerGroupState> result = ConsumerGroupCommand.consumerGroupStatesFromString("Stable");
+ assertEquals(set(Collections.singleton(ConsumerGroupState.STABLE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("Stable, PreparingRebalance");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.STABLE, ConsumerGroupState.PREPARING_REBALANCE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("Dead,CompletingRebalance,");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.DEAD, ConsumerGroupState.COMPLETING_REBALANCE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("stable");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.STABLE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("stable, assigning");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.STABLE, ConsumerGroupState.ASSIGNING)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("dead,reconciling,");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.DEAD, ConsumerGroupState.RECONCILING)), result);
+
+ assertThrows(IllegalArgumentException.class, () -> ConsumerGroupCommand.consumerGroupStatesFromString("bad, wrong"));
+
+ assertThrows(IllegalArgumentException.class, () -> ConsumerGroupCommand.consumerGroupStatesFromString(" bad, Stable"));
+
+ assertThrows(IllegalArgumentException.class, () -> ConsumerGroupCommand.consumerGroupStatesFromString(" , ,"));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListGroupCommand(String quorum) throws Exception {
+ String simpleGroup = "simple-group";
+ addSimpleGroupExecutor(simpleGroup);
+ addConsumerGroupExecutor(1);
+ final String[] out = {""};
Review Comment:
`out` value changed in java lambda
`out = grabConsoleOutput()` and only final variable (which can't be changed) can be used inside java lambda.
I can rework String array to usage of `AtomicReference`.
What do you think?
--
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-14589 [1/3] Tests of ConsoleGroupCommand rewritten in java [kafka]
Posted by "nizhikov (via GitHub)" <gi...@apache.org>.
nizhikov commented on PR #15256:
URL: https://github.com/apache/kafka/pull/15256#issuecomment-1920748368
Hello @showuon
CI is OK for me. Are you ready to merge this PR or have other questions, comments?
--
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-14589 [1/3] Tests of ConsoleGroupCommand rewritten in java [kafka]
Posted by "nizhikov (via GitHub)" <gi...@apache.org>.
nizhikov commented on code in PR #15256:
URL: https://github.com/apache/kafka/pull/15256#discussion_r1471200818
##########
tools/src/test/java/org/apache/kafka/tools/consumer/group/DeleteConsumerGroupsTest.java:
##########
@@ -0,0 +1,317 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.tools.consumer.group;
+
+import joptsimple.OptionException;
+import kafka.admin.ConsumerGroupCommand;
+import org.apache.kafka.clients.consumer.RangeAssignor;
+import org.apache.kafka.common.errors.GroupIdNotFoundException;
+import org.apache.kafka.common.errors.GroupNotEmptyException;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.test.TestUtils;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.kafka.tools.ToolsTestUtils.TEST_WITH_PARAMETERIZED_QUORUM_NAME;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class DeleteConsumerGroupsTest extends ConsumerGroupCommandTest {
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteWithTopicOption(String quorum) {
+ createOffsetsTopic(listenerName(), new Properties());
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", GROUP, "--topic"};
+ assertThrows(OptionException.class, () -> getConsumerGroupService(cgcArgs));
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteCmdNonExistingGroup(String quorum) {
+ createOffsetsTopic(listenerName(), new Properties());
+ String missingGroup = "missing.group";
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", missingGroup};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ String output = kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ service.deleteGroups();
+ return null;
+ });
+ assertTrue(output.contains("Group '" + missingGroup + "' could not be deleted due to:") && output.contains(Errors.GROUP_ID_NOT_FOUND.message()),
+ "The expected error (" + Errors.GROUP_ID_NOT_FOUND + ") was not detected while deleting consumer group");
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteNonExistingGroup(String quorum) {
+ createOffsetsTopic(listenerName(), new Properties());
+ String missingGroup = "missing.group";
+
+ // note the group to be deleted is a different (non-existing) group
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", missingGroup};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ scala.collection.Map<String, Throwable> result = service.deleteGroups();
+ assertTrue(result.size() == 1 && result.contains(missingGroup) && result.get(missingGroup).get().getCause() instanceof GroupIdNotFoundException,
+ "The expected error (" + Errors.GROUP_ID_NOT_FOUND + ") was not detected while deleting consumer group");
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteCmdNonEmptyGroup(String quorum) throws Exception {
+ createOffsetsTopic(listenerName(), new Properties());
+
+ // run one consumer in the group
+ addConsumerGroupExecutor(1);
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", GROUP};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ TestUtils.waitForCondition(
+ () -> service.collectGroupMembers(GROUP, false)._2.get().size() == 1,
+ "The group did not initialize as expected."
+ );
+
+ String output = kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ service.deleteGroups();
+ return null;
+ });
+ assertTrue(output.contains("Group '" + GROUP + "' could not be deleted due to:") && output.contains(Errors.NON_EMPTY_GROUP.message()),
+ "The expected error (" + Errors.NON_EMPTY_GROUP + ") was not detected while deleting consumer group. Output was: (" + output + ")");
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteNonEmptyGroup(String quorum) throws Exception {
+ createOffsetsTopic(listenerName(), new Properties());
+
+ // run one consumer in the group
+ addConsumerGroupExecutor(1);
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", GROUP};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ TestUtils.waitForCondition(
+ () -> service.collectGroupMembers(GROUP, false)._2.get().size() == 1,
+ "The group did not initialize as expected."
+ );
+
+ scala.collection.Map<String, Throwable> result = service.deleteGroups();
+ assertNotNull(result.get(GROUP).get(),
+ "Group was deleted successfully, but it shouldn't have been. Result was:(" + result + ")");
+ assertTrue(result.size() == 1 && result.contains(GROUP) && result.get(GROUP).get().getCause() instanceof GroupNotEmptyException,
+ "The expected error (" + Errors.NON_EMPTY_GROUP + ") was not detected while deleting consumer group. Result was:(" + result + ")");
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteCmdEmptyGroup(String quorum) throws Exception {
+ createOffsetsTopic(listenerName(), new Properties());
+
+ // run one consumer in the group
+ ConsumerGroupExecutor executor = addConsumerGroupExecutor(1);
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", GROUP};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ TestUtils.waitForCondition(
+ () -> service.listConsumerGroups().contains(GROUP) && Objects.equals(service.collectGroupState(GROUP).state(), "Stable"),
+ "The group did not initialize as expected."
+ );
+
+ executor.shutdown();
+
+ TestUtils.waitForCondition(
+ () -> Objects.equals(service.collectGroupState(GROUP).state(), "Empty"),
+ "The group did not become empty as expected."
+ );
+
+ String output = kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ service.deleteGroups();
+ return null;
+ });
+ assertTrue(output.contains("Deletion of requested consumer groups ('" + GROUP + "') was successful."),
+ "The consumer group could not be deleted as expected");
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteCmdAllGroups(String quorum) throws Exception {
+ createOffsetsTopic(listenerName(), new Properties());
+
+ // Create 3 groups with 1 consumer per each
+ Map<String, ConsumerGroupExecutor> groups = IntStream.rangeClosed(1, 3).mapToObj(i -> GROUP + i).collect(Collectors.toMap(
+ Function.identity(),
+ group -> addConsumerGroupExecutor(1, TOPIC, group, RangeAssignor.class.getName(), Optional.empty(), false)
+ ));
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--all-groups"};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ TestUtils.waitForCondition(() ->
+ Objects.equals(service.listConsumerGroups().toSet(), set(groups.keySet())) &&
+ groups.keySet().stream().allMatch(groupId -> {
+ try {
+ return Objects.equals(service.collectGroupState(groupId).state(), "Stable");
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }),
+ "The group did not initialize as expected.");
+
+ // Shutdown consumers to empty out groups
+ groups.values().forEach(AbstractConsumerGroupExecutor::shutdown);
+
+ TestUtils.waitForCondition(() ->
+ groups.keySet().stream().allMatch(groupId -> {
+ try {
+ return Objects.equals(service.collectGroupState(groupId).state(), "Empty");
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }),
+ "The group did not become empty as expected.");
+
+ String output = kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ service.deleteGroups();
+ return null;
+ }).trim();
+ Set<String> expectedGroupsForDeletion = groups.keySet();
+ Set<String> deletedGroupsGrepped = Arrays.stream(output.substring(output.indexOf('(') + 1, output.indexOf(')')).split(","))
+ .map(str -> str.replaceAll("'", "").trim()).collect(Collectors.toSet());
+
+ assertTrue(output.matches("Deletion of requested consumer groups (.*) was successful.")
+ && Objects.equals(deletedGroupsGrepped, expectedGroupsForDeletion),
+ "The consumer group(s) could not be deleted as expected");
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteEmptyGroup(String quorum) throws Exception {
+ createOffsetsTopic(listenerName(), new Properties());
+
+ // run one consumer in the group
+ ConsumerGroupExecutor executor = addConsumerGroupExecutor(1);
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", GROUP};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ TestUtils.waitForCondition(
+ () -> service.listConsumerGroups().contains(GROUP) && Objects.equals(service.collectGroupState(GROUP).state(), "Stable"),
+ "The group did not initialize as expected.");
+
+ executor.shutdown();
+
+ TestUtils.waitForCondition(
+ () -> Objects.equals(service.collectGroupState(GROUP).state(), "Empty"),
+ "The group did not become empty as expected.");
+
+ scala.collection.Map<String, Throwable> result = service.deleteGroups();
+ assertTrue(result.size() == 1 && result.contains(GROUP) && result.get(GROUP).get() == null,
+ "The consumer group could not be deleted as expected");
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteCmdWithMixOfSuccessAndError(String quorum) throws Exception {
+ createOffsetsTopic(listenerName(), new Properties());
+ String missingGroup = "missing.group";
+
+ // run one consumer in the group
+ ConsumerGroupExecutor executor = addConsumerGroupExecutor(1);
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", GROUP};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ TestUtils.waitForCondition(
+ () -> service.listConsumerGroups().contains(GROUP) && Objects.equals(service.collectGroupState(GROUP).state(), "Stable"),
+ "The group did not initialize as expected.");
+
+ executor.shutdown();
+
+ TestUtils.waitForCondition(
+ () -> Objects.equals(service.collectGroupState(GROUP).state(), "Empty"),
+ "The group did not become empty as expected.");
+
+ String[] cgcArgs2 = new String[cgcArgs.length + 2];
+
+ System.arraycopy(cgcArgs, 0, cgcArgs2, 0, cgcArgs.length);
+ cgcArgs2[cgcArgs2.length - 2] = "--group";
+ cgcArgs2[cgcArgs2.length - 1] = missingGroup;
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-14589 [1/3] Tests of ConsoleGroupCommand rewritten in java [kafka]
Posted by "nizhikov (via GitHub)" <gi...@apache.org>.
nizhikov commented on PR #15256:
URL: https://github.com/apache/kafka/pull/15256#issuecomment-1921473385
@showuon CI results are ready.
AFAICS other tests of `DynamicBrokerReconfigurationTest` failed in the latest trunk:
https://ci-builds.apache.org/blue/organizations/jenkins/Kafka%2Fkafka/detail/trunk/2609/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-14589 [1/3] Tests of ConsoleGroupCommand rewritten in java [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on PR #15256:
URL: https://github.com/apache/kafka/pull/15256#issuecomment-1911978274
Will take a look next week. Thanks.
--
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-14589 [1/3] Tests of ConsoleGroupCommand rewritten in java [kafka]
Posted by "nizhikov (via GitHub)" <gi...@apache.org>.
nizhikov commented on PR #15256:
URL: https://github.com/apache/kafka/pull/15256#issuecomment-1926418619
@showuon Other trunk run with `kafka.server.DynamicBrokerReconfigurationTest` fails - https://ci-builds.apache.org/blue/organizations/jenkins/Kafka%2Fkafka/detail/trunk/2617/tests
What do you think? Can this PR be merged as is?
--
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-14589 [1/3] Tests of ConsoleGroupCommand rewritten in java [kafka]
Posted by "nizhikov (via GitHub)" <gi...@apache.org>.
nizhikov commented on PR #15256:
URL: https://github.com/apache/kafka/pull/15256#issuecomment-1937178976
Hello @showuon
I found that `DeleteTopicTest.scala` don't close resources allocated in case of kraft run.
It was hided by removed tests and lead to the tests failures we observed earlier.
I fixed `DeleteTopicTest` and now it seems that CI is OK.
Take a look one more time, please.
--
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-14589 [1/3] Tests of ConsoleGroupCommand rewritten in java [kafka]
Posted by "nizhikov (via GitHub)" <gi...@apache.org>.
nizhikov commented on code in PR #15256:
URL: https://github.com/apache/kafka/pull/15256#discussion_r1471202618
##########
tools/src/test/java/org/apache/kafka/tools/consumer/group/DeleteConsumerGroupsTest.java:
##########
@@ -0,0 +1,317 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.tools.consumer.group;
+
+import joptsimple.OptionException;
+import kafka.admin.ConsumerGroupCommand;
+import org.apache.kafka.clients.consumer.RangeAssignor;
+import org.apache.kafka.common.errors.GroupIdNotFoundException;
+import org.apache.kafka.common.errors.GroupNotEmptyException;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.test.TestUtils;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.kafka.tools.ToolsTestUtils.TEST_WITH_PARAMETERIZED_QUORUM_NAME;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class DeleteConsumerGroupsTest extends ConsumerGroupCommandTest {
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteWithTopicOption(String quorum) {
+ createOffsetsTopic(listenerName(), new Properties());
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", GROUP, "--topic"};
+ assertThrows(OptionException.class, () -> getConsumerGroupService(cgcArgs));
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteCmdNonExistingGroup(String quorum) {
+ createOffsetsTopic(listenerName(), new Properties());
+ String missingGroup = "missing.group";
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", missingGroup};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ String output = kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ service.deleteGroups();
+ return null;
+ });
+ assertTrue(output.contains("Group '" + missingGroup + "' could not be deleted due to:") && output.contains(Errors.GROUP_ID_NOT_FOUND.message()),
+ "The expected error (" + Errors.GROUP_ID_NOT_FOUND + ") was not detected while deleting consumer group");
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteNonExistingGroup(String quorum) {
+ createOffsetsTopic(listenerName(), new Properties());
+ String missingGroup = "missing.group";
+
+ // note the group to be deleted is a different (non-existing) group
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", missingGroup};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ scala.collection.Map<String, Throwable> result = service.deleteGroups();
+ assertTrue(result.size() == 1 && result.contains(missingGroup) && result.get(missingGroup).get().getCause() instanceof GroupIdNotFoundException,
+ "The expected error (" + Errors.GROUP_ID_NOT_FOUND + ") was not detected while deleting consumer group");
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteCmdNonEmptyGroup(String quorum) throws Exception {
+ createOffsetsTopic(listenerName(), new Properties());
+
+ // run one consumer in the group
+ addConsumerGroupExecutor(1);
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", GROUP};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ TestUtils.waitForCondition(
+ () -> service.collectGroupMembers(GROUP, false)._2.get().size() == 1,
+ "The group did not initialize as expected."
+ );
+
+ String output = kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ service.deleteGroups();
+ return null;
+ });
+ assertTrue(output.contains("Group '" + GROUP + "' could not be deleted due to:") && output.contains(Errors.NON_EMPTY_GROUP.message()),
+ "The expected error (" + Errors.NON_EMPTY_GROUP + ") was not detected while deleting consumer group. Output was: (" + output + ")");
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteNonEmptyGroup(String quorum) throws Exception {
+ createOffsetsTopic(listenerName(), new Properties());
+
+ // run one consumer in the group
+ addConsumerGroupExecutor(1);
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", GROUP};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ TestUtils.waitForCondition(
+ () -> service.collectGroupMembers(GROUP, false)._2.get().size() == 1,
+ "The group did not initialize as expected."
+ );
+
+ scala.collection.Map<String, Throwable> result = service.deleteGroups();
+ assertNotNull(result.get(GROUP).get(),
+ "Group was deleted successfully, but it shouldn't have been. Result was:(" + result + ")");
+ assertTrue(result.size() == 1 && result.contains(GROUP) && result.get(GROUP).get().getCause() instanceof GroupNotEmptyException,
+ "The expected error (" + Errors.NON_EMPTY_GROUP + ") was not detected while deleting consumer group. Result was:(" + result + ")");
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteCmdEmptyGroup(String quorum) throws Exception {
+ createOffsetsTopic(listenerName(), new Properties());
+
+ // run one consumer in the group
+ ConsumerGroupExecutor executor = addConsumerGroupExecutor(1);
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", GROUP};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ TestUtils.waitForCondition(
+ () -> service.listConsumerGroups().contains(GROUP) && Objects.equals(service.collectGroupState(GROUP).state(), "Stable"),
+ "The group did not initialize as expected."
+ );
+
+ executor.shutdown();
+
+ TestUtils.waitForCondition(
+ () -> Objects.equals(service.collectGroupState(GROUP).state(), "Empty"),
+ "The group did not become empty as expected."
+ );
+
+ String output = kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ service.deleteGroups();
+ return null;
+ });
+ assertTrue(output.contains("Deletion of requested consumer groups ('" + GROUP + "') was successful."),
+ "The consumer group could not be deleted as expected");
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteCmdAllGroups(String quorum) throws Exception {
+ createOffsetsTopic(listenerName(), new Properties());
+
+ // Create 3 groups with 1 consumer per each
+ Map<String, ConsumerGroupExecutor> groups = IntStream.rangeClosed(1, 3).mapToObj(i -> GROUP + i).collect(Collectors.toMap(
+ Function.identity(),
+ group -> addConsumerGroupExecutor(1, TOPIC, group, RangeAssignor.class.getName(), Optional.empty(), false)
+ ));
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--all-groups"};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ TestUtils.waitForCondition(() ->
+ Objects.equals(service.listConsumerGroups().toSet(), set(groups.keySet())) &&
+ groups.keySet().stream().allMatch(groupId -> {
+ try {
+ return Objects.equals(service.collectGroupState(groupId).state(), "Stable");
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }),
+ "The group did not initialize as expected.");
+
+ // Shutdown consumers to empty out groups
+ groups.values().forEach(AbstractConsumerGroupExecutor::shutdown);
+
+ TestUtils.waitForCondition(() ->
+ groups.keySet().stream().allMatch(groupId -> {
+ try {
+ return Objects.equals(service.collectGroupState(groupId).state(), "Empty");
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }),
+ "The group did not become empty as expected.");
+
+ String output = kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ service.deleteGroups();
+ return null;
+ }).trim();
+ Set<String> expectedGroupsForDeletion = groups.keySet();
+ Set<String> deletedGroupsGrepped = Arrays.stream(output.substring(output.indexOf('(') + 1, output.indexOf(')')).split(","))
+ .map(str -> str.replaceAll("'", "").trim()).collect(Collectors.toSet());
+
+ assertTrue(output.matches("Deletion of requested consumer groups (.*) was successful.")
+ && Objects.equals(deletedGroupsGrepped, expectedGroupsForDeletion),
+ "The consumer group(s) could not be deleted as expected");
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteEmptyGroup(String quorum) throws Exception {
+ createOffsetsTopic(listenerName(), new Properties());
+
+ // run one consumer in the group
+ ConsumerGroupExecutor executor = addConsumerGroupExecutor(1);
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", GROUP};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ TestUtils.waitForCondition(
+ () -> service.listConsumerGroups().contains(GROUP) && Objects.equals(service.collectGroupState(GROUP).state(), "Stable"),
+ "The group did not initialize as expected.");
+
+ executor.shutdown();
+
+ TestUtils.waitForCondition(
+ () -> Objects.equals(service.collectGroupState(GROUP).state(), "Empty"),
+ "The group did not become empty as expected.");
+
+ scala.collection.Map<String, Throwable> result = service.deleteGroups();
+ assertTrue(result.size() == 1 && result.contains(GROUP) && result.get(GROUP).get() == null,
+ "The consumer group could not be deleted as expected");
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteCmdWithMixOfSuccessAndError(String quorum) throws Exception {
+ createOffsetsTopic(listenerName(), new Properties());
+ String missingGroup = "missing.group";
+
+ // run one consumer in the group
+ ConsumerGroupExecutor executor = addConsumerGroupExecutor(1);
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", GROUP};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ TestUtils.waitForCondition(
+ () -> service.listConsumerGroups().contains(GROUP) && Objects.equals(service.collectGroupState(GROUP).state(), "Stable"),
+ "The group did not initialize as expected.");
+
+ executor.shutdown();
+
+ TestUtils.waitForCondition(
+ () -> Objects.equals(service.collectGroupState(GROUP).state(), "Empty"),
+ "The group did not become empty as expected.");
+
+ String[] cgcArgs2 = new String[cgcArgs.length + 2];
+
+ System.arraycopy(cgcArgs, 0, cgcArgs2, 0, cgcArgs.length);
+ cgcArgs2[cgcArgs2.length - 2] = "--group";
+ cgcArgs2[cgcArgs2.length - 1] = missingGroup;
+
+ ConsumerGroupCommand.ConsumerGroupService service2 = getConsumerGroupService(cgcArgs2);
+
+ String output = kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ service2.deleteGroups();
+ return null;
+ });
+ assertTrue(output.contains("Group '" + missingGroup + "' could not be deleted due to:")
+ && output.contains(Errors.GROUP_ID_NOT_FOUND.message())
+ && output.contains("These consumer groups were deleted successfully: '" + GROUP + "'"),
+ "The consumer group deletion did not work as expected");
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteWithMixOfSuccessAndError(String quorum) throws Exception {
+ createOffsetsTopic(listenerName(), new Properties());
+ String missingGroup = "missing.group";
+
+ // run one consumer in the group
+ ConsumerGroupExecutor executor = addConsumerGroupExecutor(1);
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", GROUP};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ TestUtils.waitForCondition(
+ () -> service.listConsumerGroups().contains(GROUP) && Objects.equals(service.collectGroupState(GROUP).state(), "Stable"),
+ "The group did not initialize as expected.");
+
+ executor.shutdown();
+
+ TestUtils.waitForCondition(
+ () -> Objects.equals(service.collectGroupState(GROUP).state(), "Empty"),
+ "The group did not become empty as expected.");
+
+ String[] cgcArgs2 = new String[cgcArgs.length + 2];
+
+ System.arraycopy(cgcArgs, 0, cgcArgs2, 0, cgcArgs.length);
+ cgcArgs2[cgcArgs2.length - 2] = "--group";
+ cgcArgs2[cgcArgs2.length - 1] = missingGroup;
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-14589 [1/3] Tests of ConsoleGroupCommand rewritten in java [kafka]
Posted by "nizhikov (via GitHub)" <gi...@apache.org>.
nizhikov commented on code in PR #15256:
URL: https://github.com/apache/kafka/pull/15256#discussion_r1471221535
##########
tools/src/test/java/org/apache/kafka/tools/consumer/group/ListConsumerGroupTest.java:
##########
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.tools.consumer.group;
+
+import joptsimple.OptionException;
+import kafka.admin.ConsumerGroupCommand;
+import org.apache.kafka.clients.admin.ConsumerGroupListing;
+import org.apache.kafka.common.ConsumerGroupState;
+import org.apache.kafka.test.TestUtils;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Objects;
+import java.util.Optional;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class ListConsumerGroupTest extends ConsumerGroupCommandTest {
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListConsumerGroups(String quorum) throws Exception {
+ String simpleGroup = "simple-group";
+ addSimpleGroupExecutor(simpleGroup);
+ addConsumerGroupExecutor(1);
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list"};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+ scala.collection.Set<String> expectedGroups = set(Arrays.asList(GROUP, simpleGroup));
+ final scala.collection.Set[] foundGroups = new scala.collection.Set[1];
+ TestUtils.waitForCondition(() -> {
+ foundGroups[0] = service.listConsumerGroups().toSet();
+ return Objects.equals(expectedGroups, foundGroups[0]);
+ }, "Expected --list to show groups " + expectedGroups + ", but found " + foundGroups[0] + ".");
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListWithUnrecognizedNewConsumerOption() {
+ String[] cgcArgs = new String[]{"--new-consumer", "--bootstrap-server", bootstrapServers(listenerName()), "--list"};
+ assertThrows(OptionException.class, () -> getConsumerGroupService(cgcArgs));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListConsumerGroupsWithStates() throws Exception {
+ String simpleGroup = "simple-group";
+ addSimpleGroupExecutor(simpleGroup);
+ addConsumerGroupExecutor(1);
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list", "--state"};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ scala.collection.Set<ConsumerGroupListing> expectedListing = set(Arrays.asList(
+ new ConsumerGroupListing(simpleGroup, true, Optional.of(ConsumerGroupState.EMPTY)),
+ new ConsumerGroupListing(GROUP, false, Optional.of(ConsumerGroupState.STABLE))));
+
+ final scala.collection.Set[] foundListing = new scala.collection.Set[1];
Review Comment:
It seems that usage of scala collections is more convinient.
But, of course, when java version of command will be merged we will use java collections here and everywhere in test. You can see how this don for already mereged tests:
https://github.com/apache/kafka/pull/14471/files#diff-4e22e01d567ea1b5087aa60354fa127bb25b892b430a6b86ab39a10edd393c0a
--
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-14589 [1/3] Tests of ConsoleGroupCommand rewritten in java [kafka]
Posted by "nizhikov (via GitHub)" <gi...@apache.org>.
nizhikov commented on PR #15256:
URL: https://github.com/apache/kafka/pull/15256#issuecomment-1920969424
@dajac It has impact, for sure :)
I reflect your changes in the main PR #14471 and will merge them to this PR in the nearest time.
--
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-14589 [1/3] Tests of ConsoleGroupCommand rewritten in java [kafka]
Posted by "nizhikov (via GitHub)" <gi...@apache.org>.
nizhikov commented on PR #15256:
URL: https://github.com/apache/kafka/pull/15256#issuecomment-1921875382
@showuon Please, wright me, If those failures should be further investigated.
--
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-14589 [1/3] Tests of ConsoleGroupCommand rewritten in java [kafka]
Posted by "nizhikov (via GitHub)" <gi...@apache.org>.
nizhikov commented on PR #15256:
URL: https://github.com/apache/kafka/pull/15256#issuecomment-1925708621
@showuon Looks like `kafka.server.DynamicBrokerReconfigurationTest.testThreadPoolResize()` fails in the latest trunk, also:
https://ci-builds.apache.org/blue/organizations/jenkins/Kafka%2Fkafka/detail/trunk/2618/tests
So, maybe the fail not caused by this PR?
--
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-14589 [1/3] Tests of ConsoleGroupCommand rewritten in java [kafka]
Posted by "nizhikov (via GitHub)" <gi...@apache.org>.
nizhikov commented on code in PR #15256:
URL: https://github.com/apache/kafka/pull/15256#discussion_r1471180248
##########
tools/src/test/java/org/apache/kafka/tools/consumer/group/DeleteConsumerGroupsTest.java:
##########
@@ -0,0 +1,317 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.tools.consumer.group;
+
+import joptsimple.OptionException;
+import kafka.admin.ConsumerGroupCommand;
+import org.apache.kafka.clients.consumer.RangeAssignor;
+import org.apache.kafka.common.errors.GroupIdNotFoundException;
+import org.apache.kafka.common.errors.GroupNotEmptyException;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.test.TestUtils;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.kafka.tools.ToolsTestUtils.TEST_WITH_PARAMETERIZED_QUORUM_NAME;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class DeleteConsumerGroupsTest extends ConsumerGroupCommandTest {
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteWithTopicOption(String quorum) {
+ createOffsetsTopic(listenerName(), new Properties());
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", GROUP, "--topic"};
+ assertThrows(OptionException.class, () -> getConsumerGroupService(cgcArgs));
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteCmdNonExistingGroup(String quorum) {
+ createOffsetsTopic(listenerName(), new Properties());
+ String missingGroup = "missing.group";
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", missingGroup};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ String output = kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ service.deleteGroups();
+ return null;
+ });
+ assertTrue(output.contains("Group '" + missingGroup + "' could not be deleted due to:") && output.contains(Errors.GROUP_ID_NOT_FOUND.message()),
+ "The expected error (" + Errors.GROUP_ID_NOT_FOUND + ") was not detected while deleting consumer group");
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteNonExistingGroup(String quorum) {
+ createOffsetsTopic(listenerName(), new Properties());
+ String missingGroup = "missing.group";
+
+ // note the group to be deleted is a different (non-existing) group
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", missingGroup};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ scala.collection.Map<String, Throwable> result = service.deleteGroups();
+ assertTrue(result.size() == 1 && result.contains(missingGroup) && result.get(missingGroup).get().getCause() instanceof GroupIdNotFoundException,
+ "The expected error (" + Errors.GROUP_ID_NOT_FOUND + ") was not detected while deleting consumer group");
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteCmdNonEmptyGroup(String quorum) throws Exception {
+ createOffsetsTopic(listenerName(), new Properties());
+
+ // run one consumer in the group
+ addConsumerGroupExecutor(1);
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", GROUP};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ TestUtils.waitForCondition(
+ () -> service.collectGroupMembers(GROUP, false)._2.get().size() == 1,
+ "The group did not initialize as expected."
+ );
+
+ String output = kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ service.deleteGroups();
+ return null;
Review Comment:
Yes.
--
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-14589 [1/3] Tests of ConsoleGroupCommand rewritten in java [kafka]
Posted by "nizhikov (via GitHub)" <gi...@apache.org>.
nizhikov commented on code in PR #15256:
URL: https://github.com/apache/kafka/pull/15256#discussion_r1471191734
##########
tools/src/test/java/org/apache/kafka/tools/consumer/group/DeleteOffsetsConsumerGroupCommandIntegrationTest.java:
##########
@@ -0,0 +1,205 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.tools.consumer.group;
+
+import kafka.admin.ConsumerGroupCommand;
+import kafka.utils.TestUtils;
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.server.config.Defaults;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.Properties;
+import java.util.concurrent.ExecutionException;
+
+import static org.apache.kafka.test.TestUtils.DEFAULT_MAX_WAIT_MS;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+public class DeleteOffsetsConsumerGroupCommandIntegrationTest extends ConsumerGroupCommandTest {
+ String[] getArgs(String group, String topic) {
+ return new String[] {
+ "--bootstrap-server", bootstrapServers(listenerName()),
+ "--delete-offsets",
+ "--group", group,
+ "--topic", topic
+ };
+ }
+
+ @Test
Review Comment:
Great catch. Thanks. Fixed.
This was added 2 weeks ago and I missed to reflect that change in java test version.
--
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-14589 [1/3] Tests of ConsoleGroupCommand rewritten in java [kafka]
Posted by "nizhikov (via GitHub)" <gi...@apache.org>.
nizhikov commented on code in PR #15256:
URL: https://github.com/apache/kafka/pull/15256#discussion_r1471245590
##########
tools/src/test/java/org/apache/kafka/tools/consumer/group/ListConsumerGroupTest.java:
##########
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.tools.consumer.group;
+
+import joptsimple.OptionException;
+import kafka.admin.ConsumerGroupCommand;
+import org.apache.kafka.clients.admin.ConsumerGroupListing;
+import org.apache.kafka.common.ConsumerGroupState;
+import org.apache.kafka.test.TestUtils;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Objects;
+import java.util.Optional;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class ListConsumerGroupTest extends ConsumerGroupCommandTest {
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListConsumerGroups(String quorum) throws Exception {
+ String simpleGroup = "simple-group";
+ addSimpleGroupExecutor(simpleGroup);
+ addConsumerGroupExecutor(1);
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list"};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+ scala.collection.Set<String> expectedGroups = set(Arrays.asList(GROUP, simpleGroup));
+ final scala.collection.Set[] foundGroups = new scala.collection.Set[1];
+ TestUtils.waitForCondition(() -> {
+ foundGroups[0] = service.listConsumerGroups().toSet();
+ return Objects.equals(expectedGroups, foundGroups[0]);
+ }, "Expected --list to show groups " + expectedGroups + ", but found " + foundGroups[0] + ".");
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListWithUnrecognizedNewConsumerOption() {
+ String[] cgcArgs = new String[]{"--new-consumer", "--bootstrap-server", bootstrapServers(listenerName()), "--list"};
+ assertThrows(OptionException.class, () -> getConsumerGroupService(cgcArgs));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListConsumerGroupsWithStates() throws Exception {
+ String simpleGroup = "simple-group";
+ addSimpleGroupExecutor(simpleGroup);
+ addConsumerGroupExecutor(1);
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list", "--state"};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ scala.collection.Set<ConsumerGroupListing> expectedListing = set(Arrays.asList(
+ new ConsumerGroupListing(simpleGroup, true, Optional.of(ConsumerGroupState.EMPTY)),
+ new ConsumerGroupListing(GROUP, false, Optional.of(ConsumerGroupState.STABLE))));
+
+ final scala.collection.Set[] foundListing = new scala.collection.Set[1];
+ TestUtils.waitForCondition(() -> {
+ foundListing[0] = service.listConsumerGroupsWithState(set(Arrays.asList(ConsumerGroupState.values()))).toSet();
+ return Objects.equals(expectedListing, foundListing[0]);
+ }, "Expected to show groups " + expectedListing + ", but found " + foundListing[0]);
+
+ scala.collection.Set<ConsumerGroupListing> expectedListingStable = set(Collections.singleton(
+ new ConsumerGroupListing(GROUP, false, Optional.of(ConsumerGroupState.STABLE))));
+
+ foundListing[0] = null;
+
+ TestUtils.waitForCondition(() -> {
+ foundListing[0] = service.listConsumerGroupsWithState(set(Collections.singleton(ConsumerGroupState.STABLE))).toSet();
+ return Objects.equals(expectedListingStable, foundListing[0]);
+ }, "Expected to show groups " + expectedListingStable + ", but found " + foundListing[0]);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testConsumerGroupStatesFromString(String quorum) {
+ scala.collection.Set<ConsumerGroupState> result = ConsumerGroupCommand.consumerGroupStatesFromString("Stable");
+ assertEquals(set(Collections.singleton(ConsumerGroupState.STABLE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("Stable, PreparingRebalance");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.STABLE, ConsumerGroupState.PREPARING_REBALANCE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("Dead,CompletingRebalance,");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.DEAD, ConsumerGroupState.COMPLETING_REBALANCE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("stable");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.STABLE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("stable, assigning");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.STABLE, ConsumerGroupState.ASSIGNING)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("dead,reconciling,");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.DEAD, ConsumerGroupState.RECONCILING)), result);
+
+ assertThrows(IllegalArgumentException.class, () -> ConsumerGroupCommand.consumerGroupStatesFromString("bad, wrong"));
+
+ assertThrows(IllegalArgumentException.class, () -> ConsumerGroupCommand.consumerGroupStatesFromString(" bad, Stable"));
+
+ assertThrows(IllegalArgumentException.class, () -> ConsumerGroupCommand.consumerGroupStatesFromString(" , ,"));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListGroupCommand(String quorum) throws Exception {
+ String simpleGroup = "simple-group";
+ addSimpleGroupExecutor(simpleGroup);
+ addConsumerGroupExecutor(1);
+ final String[] out = {""};
Review Comment:
Reworked to `AtomicReference`.
--
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-14589 [1/3] Tests of ConsoleGroupCommand rewritten in java [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on PR #15256:
URL: https://github.com/apache/kafka/pull/15256#issuecomment-1923368577
I think the `kafka.server.DynamicBrokerReconfigurationTest.testThreadPoolResize()` is caused by this PR. We need to investigate it. From the error, it said there is a thread leaked:
```
org.opentest4j.AssertionFailedError: Invalid threads: expected 6, got 7: List(data-plane-kafka-socket-acceptor-ListenerName(EXTERNAL)-SASL_SSL-0, data-plane-kafka-socket-acceptor-ListenerName(INTERNAL)-SSL-0, data-plane-kafka-socket-acceptor-ListenerName(CONTROLLER)-PLAINTEXT-0, data-plane-kafka-socket-acceptor-ListenerName(INTERNAL)-SSL-0, data-plane-kafka-socket-acceptor-ListenerName(INTERNAL)-SSL-0, data-plane-kafka-socket-acceptor-ListenerName(EXTERNAL)-SASL_SSL-0, data-plane-kafka-socket-acceptor-ListenerName(EXTERNAL)-SASL_SSL-0) ==> expected: <true> but was: <false>
```
You can try to run your changed test suites 1 by 1, and check that after all test suite completes, is there any network thread leaked. Let me know if you need help.
--
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-14589 [1/3] Tests of ConsoleGroupCommand rewritten in java [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on PR #15256:
URL: https://github.com/apache/kafka/pull/15256#issuecomment-1926766135
I just re-run CI this morning, and this time, it failed 2 tests in all 4 jobs. Trying to run it again. I can't merge anything introducing test failures, I think you also understand it. It would be helpful if you could help identify the reason why it failed.
```
Build / JDK 8 and Scala 2.12 / kafka.api.SslAdminIntegrationTest.testAsynchronousAuthorizerAclUpdatesDontBlockRequestThreads()
Build / JDK 8 and Scala 2.12 / kafka.api.SslAdminIntegrationTest.testSynchronousAuthorizerAclUpdatesBlockRequestThreads()
```
--
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-14589 [1/3] Tests of ConsoleGroupCommand rewritten in java [kafka]
Posted by "nizhikov (via GitHub)" <gi...@apache.org>.
nizhikov commented on PR #15256:
URL: https://github.com/apache/kafka/pull/15256#issuecomment-1918995072
CI results looks good for me.
--
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-14589 [1/3] Tests of ConsoleGroupCommand rewritten in java [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon merged PR #15256:
URL: https://github.com/apache/kafka/pull/15256
--
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-14589 [1/3] Tests of ConsoleGroupCommand rewritten in java [kafka]
Posted by "nizhikov (via GitHub)" <gi...@apache.org>.
nizhikov commented on code in PR #15256:
URL: https://github.com/apache/kafka/pull/15256#discussion_r1471241531
##########
tools/src/test/java/org/apache/kafka/tools/consumer/group/ListConsumerGroupTest.java:
##########
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.tools.consumer.group;
+
+import joptsimple.OptionException;
+import kafka.admin.ConsumerGroupCommand;
+import org.apache.kafka.clients.admin.ConsumerGroupListing;
+import org.apache.kafka.common.ConsumerGroupState;
+import org.apache.kafka.test.TestUtils;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Objects;
+import java.util.Optional;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class ListConsumerGroupTest extends ConsumerGroupCommandTest {
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListConsumerGroups(String quorum) throws Exception {
+ String simpleGroup = "simple-group";
+ addSimpleGroupExecutor(simpleGroup);
+ addConsumerGroupExecutor(1);
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list"};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+ scala.collection.Set<String> expectedGroups = set(Arrays.asList(GROUP, simpleGroup));
+ final scala.collection.Set[] foundGroups = new scala.collection.Set[1];
+ TestUtils.waitForCondition(() -> {
+ foundGroups[0] = service.listConsumerGroups().toSet();
+ return Objects.equals(expectedGroups, foundGroups[0]);
+ }, "Expected --list to show groups " + expectedGroups + ", but found " + foundGroups[0] + ".");
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListWithUnrecognizedNewConsumerOption() {
+ String[] cgcArgs = new String[]{"--new-consumer", "--bootstrap-server", bootstrapServers(listenerName()), "--list"};
+ assertThrows(OptionException.class, () -> getConsumerGroupService(cgcArgs));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListConsumerGroupsWithStates() throws Exception {
+ String simpleGroup = "simple-group";
+ addSimpleGroupExecutor(simpleGroup);
+ addConsumerGroupExecutor(1);
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list", "--state"};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ scala.collection.Set<ConsumerGroupListing> expectedListing = set(Arrays.asList(
+ new ConsumerGroupListing(simpleGroup, true, Optional.of(ConsumerGroupState.EMPTY)),
+ new ConsumerGroupListing(GROUP, false, Optional.of(ConsumerGroupState.STABLE))));
+
+ final scala.collection.Set[] foundListing = new scala.collection.Set[1];
+ TestUtils.waitForCondition(() -> {
+ foundListing[0] = service.listConsumerGroupsWithState(set(Arrays.asList(ConsumerGroupState.values()))).toSet();
+ return Objects.equals(expectedListing, foundListing[0]);
+ }, "Expected to show groups " + expectedListing + ", but found " + foundListing[0]);
+
+ scala.collection.Set<ConsumerGroupListing> expectedListingStable = set(Collections.singleton(
+ new ConsumerGroupListing(GROUP, false, Optional.of(ConsumerGroupState.STABLE))));
+
+ foundListing[0] = null;
+
+ TestUtils.waitForCondition(() -> {
+ foundListing[0] = service.listConsumerGroupsWithState(set(Collections.singleton(ConsumerGroupState.STABLE))).toSet();
+ return Objects.equals(expectedListingStable, foundListing[0]);
+ }, "Expected to show groups " + expectedListingStable + ", but found " + foundListing[0]);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testConsumerGroupStatesFromString(String quorum) {
+ scala.collection.Set<ConsumerGroupState> result = ConsumerGroupCommand.consumerGroupStatesFromString("Stable");
+ assertEquals(set(Collections.singleton(ConsumerGroupState.STABLE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("Stable, PreparingRebalance");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.STABLE, ConsumerGroupState.PREPARING_REBALANCE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("Dead,CompletingRebalance,");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.DEAD, ConsumerGroupState.COMPLETING_REBALANCE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("stable");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.STABLE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("stable, assigning");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.STABLE, ConsumerGroupState.ASSIGNING)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("dead,reconciling,");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.DEAD, ConsumerGroupState.RECONCILING)), result);
+
+ assertThrows(IllegalArgumentException.class, () -> ConsumerGroupCommand.consumerGroupStatesFromString("bad, wrong"));
+
+ assertThrows(IllegalArgumentException.class, () -> ConsumerGroupCommand.consumerGroupStatesFromString(" bad, Stable"));
+
+ assertThrows(IllegalArgumentException.class, () -> ConsumerGroupCommand.consumerGroupStatesFromString(" , ,"));
Review Comment:
Double checked and can't spot missing test. Can you, please, clarify what test are missing?
--
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-14589 [1/3] Tests of ConsoleGroupCommand rewritten in java [kafka]
Posted by "nizhikov (via GitHub)" <gi...@apache.org>.
nizhikov commented on code in PR #15256:
URL: https://github.com/apache/kafka/pull/15256#discussion_r1471228675
##########
tools/src/test/java/org/apache/kafka/tools/consumer/group/ListConsumerGroupTest.java:
##########
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.tools.consumer.group;
+
+import joptsimple.OptionException;
+import kafka.admin.ConsumerGroupCommand;
+import org.apache.kafka.clients.admin.ConsumerGroupListing;
+import org.apache.kafka.common.ConsumerGroupState;
+import org.apache.kafka.test.TestUtils;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Objects;
+import java.util.Optional;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class ListConsumerGroupTest extends ConsumerGroupCommandTest {
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListConsumerGroups(String quorum) throws Exception {
+ String simpleGroup = "simple-group";
+ addSimpleGroupExecutor(simpleGroup);
+ addConsumerGroupExecutor(1);
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list"};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+ scala.collection.Set<String> expectedGroups = set(Arrays.asList(GROUP, simpleGroup));
+ final scala.collection.Set[] foundGroups = new scala.collection.Set[1];
+ TestUtils.waitForCondition(() -> {
+ foundGroups[0] = service.listConsumerGroups().toSet();
+ return Objects.equals(expectedGroups, foundGroups[0]);
+ }, "Expected --list to show groups " + expectedGroups + ", but found " + foundGroups[0] + ".");
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListWithUnrecognizedNewConsumerOption() {
+ String[] cgcArgs = new String[]{"--new-consumer", "--bootstrap-server", bootstrapServers(listenerName()), "--list"};
+ assertThrows(OptionException.class, () -> getConsumerGroupService(cgcArgs));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListConsumerGroupsWithStates() throws Exception {
+ String simpleGroup = "simple-group";
+ addSimpleGroupExecutor(simpleGroup);
+ addConsumerGroupExecutor(1);
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list", "--state"};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ scala.collection.Set<ConsumerGroupListing> expectedListing = set(Arrays.asList(
+ new ConsumerGroupListing(simpleGroup, true, Optional.of(ConsumerGroupState.EMPTY)),
+ new ConsumerGroupListing(GROUP, false, Optional.of(ConsumerGroupState.STABLE))));
+
+ final scala.collection.Set[] foundListing = new scala.collection.Set[1];
+ TestUtils.waitForCondition(() -> {
+ foundListing[0] = service.listConsumerGroupsWithState(set(Arrays.asList(ConsumerGroupState.values()))).toSet();
+ return Objects.equals(expectedListing, foundListing[0]);
+ }, "Expected to show groups " + expectedListing + ", but found " + foundListing[0]);
+
+ scala.collection.Set<ConsumerGroupListing> expectedListingStable = set(Collections.singleton(
+ new ConsumerGroupListing(GROUP, false, Optional.of(ConsumerGroupState.STABLE))));
+
+ foundListing[0] = null;
+
+ TestUtils.waitForCondition(() -> {
+ foundListing[0] = service.listConsumerGroupsWithState(set(Collections.singleton(ConsumerGroupState.STABLE))).toSet();
+ return Objects.equals(expectedListingStable, foundListing[0]);
+ }, "Expected to show groups " + expectedListingStable + ", but found " + foundListing[0]);
Review Comment:
Reworked with `AtomicReference`.
--
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-14589 [1/3] Tests of ConsoleGroupCommand rewritten in java [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #15256:
URL: https://github.com/apache/kafka/pull/15256#discussion_r1471122417
##########
tools/src/test/java/org/apache/kafka/tools/consumer/group/DeleteOffsetsConsumerGroupCommandIntegrationTest.java:
##########
@@ -0,0 +1,205 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.tools.consumer.group;
+
+import kafka.admin.ConsumerGroupCommand;
+import kafka.utils.TestUtils;
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.server.config.Defaults;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.Properties;
+import java.util.concurrent.ExecutionException;
+
+import static org.apache.kafka.test.TestUtils.DEFAULT_MAX_WAIT_MS;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+public class DeleteOffsetsConsumerGroupCommandIntegrationTest extends ConsumerGroupCommandTest {
+ String[] getArgs(String group, String topic) {
+ return new String[] {
+ "--bootstrap-server", bootstrapServers(listenerName()),
+ "--delete-offsets",
+ "--group", group,
+ "--topic", topic
+ };
+ }
+
+ @Test
Review Comment:
Should we add these?
```
@ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
@ValueSource(strings = {"zk", "kraft"})
```
Same for below tests.
##########
tools/src/test/java/org/apache/kafka/tools/consumer/group/ListConsumerGroupTest.java:
##########
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.tools.consumer.group;
+
+import joptsimple.OptionException;
+import kafka.admin.ConsumerGroupCommand;
+import org.apache.kafka.clients.admin.ConsumerGroupListing;
+import org.apache.kafka.common.ConsumerGroupState;
+import org.apache.kafka.test.TestUtils;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Objects;
+import java.util.Optional;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class ListConsumerGroupTest extends ConsumerGroupCommandTest {
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListConsumerGroups(String quorum) throws Exception {
+ String simpleGroup = "simple-group";
+ addSimpleGroupExecutor(simpleGroup);
+ addConsumerGroupExecutor(1);
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list"};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+ scala.collection.Set<String> expectedGroups = set(Arrays.asList(GROUP, simpleGroup));
+ final scala.collection.Set[] foundGroups = new scala.collection.Set[1];
+ TestUtils.waitForCondition(() -> {
+ foundGroups[0] = service.listConsumerGroups().toSet();
+ return Objects.equals(expectedGroups, foundGroups[0]);
+ }, "Expected --list to show groups " + expectedGroups + ", but found " + foundGroups[0] + ".");
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListWithUnrecognizedNewConsumerOption() {
+ String[] cgcArgs = new String[]{"--new-consumer", "--bootstrap-server", bootstrapServers(listenerName()), "--list"};
+ assertThrows(OptionException.class, () -> getConsumerGroupService(cgcArgs));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListConsumerGroupsWithStates() throws Exception {
+ String simpleGroup = "simple-group";
+ addSimpleGroupExecutor(simpleGroup);
+ addConsumerGroupExecutor(1);
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list", "--state"};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ scala.collection.Set<ConsumerGroupListing> expectedListing = set(Arrays.asList(
+ new ConsumerGroupListing(simpleGroup, true, Optional.of(ConsumerGroupState.EMPTY)),
+ new ConsumerGroupListing(GROUP, false, Optional.of(ConsumerGroupState.STABLE))));
+
+ final scala.collection.Set[] foundListing = new scala.collection.Set[1];
Review Comment:
Could we use java collection here?
##########
tools/src/test/java/org/apache/kafka/tools/consumer/group/ListConsumerGroupTest.java:
##########
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.tools.consumer.group;
+
+import joptsimple.OptionException;
+import kafka.admin.ConsumerGroupCommand;
+import org.apache.kafka.clients.admin.ConsumerGroupListing;
+import org.apache.kafka.common.ConsumerGroupState;
+import org.apache.kafka.test.TestUtils;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Objects;
+import java.util.Optional;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class ListConsumerGroupTest extends ConsumerGroupCommandTest {
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListConsumerGroups(String quorum) throws Exception {
+ String simpleGroup = "simple-group";
+ addSimpleGroupExecutor(simpleGroup);
+ addConsumerGroupExecutor(1);
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list"};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+ scala.collection.Set<String> expectedGroups = set(Arrays.asList(GROUP, simpleGroup));
+ final scala.collection.Set[] foundGroups = new scala.collection.Set[1];
+ TestUtils.waitForCondition(() -> {
+ foundGroups[0] = service.listConsumerGroups().toSet();
+ return Objects.equals(expectedGroups, foundGroups[0]);
+ }, "Expected --list to show groups " + expectedGroups + ", but found " + foundGroups[0] + ".");
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListWithUnrecognizedNewConsumerOption() {
+ String[] cgcArgs = new String[]{"--new-consumer", "--bootstrap-server", bootstrapServers(listenerName()), "--list"};
+ assertThrows(OptionException.class, () -> getConsumerGroupService(cgcArgs));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListConsumerGroupsWithStates() throws Exception {
+ String simpleGroup = "simple-group";
+ addSimpleGroupExecutor(simpleGroup);
+ addConsumerGroupExecutor(1);
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list", "--state"};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ scala.collection.Set<ConsumerGroupListing> expectedListing = set(Arrays.asList(
+ new ConsumerGroupListing(simpleGroup, true, Optional.of(ConsumerGroupState.EMPTY)),
+ new ConsumerGroupListing(GROUP, false, Optional.of(ConsumerGroupState.STABLE))));
+
+ final scala.collection.Set[] foundListing = new scala.collection.Set[1];
+ TestUtils.waitForCondition(() -> {
+ foundListing[0] = service.listConsumerGroupsWithState(set(Arrays.asList(ConsumerGroupState.values()))).toSet();
+ return Objects.equals(expectedListing, foundListing[0]);
+ }, "Expected to show groups " + expectedListing + ", but found " + foundListing[0]);
+
+ scala.collection.Set<ConsumerGroupListing> expectedListingStable = set(Collections.singleton(
+ new ConsumerGroupListing(GROUP, false, Optional.of(ConsumerGroupState.STABLE))));
+
+ foundListing[0] = null;
+
+ TestUtils.waitForCondition(() -> {
+ foundListing[0] = service.listConsumerGroupsWithState(set(Collections.singleton(ConsumerGroupState.STABLE))).toSet();
+ return Objects.equals(expectedListingStable, foundListing[0]);
+ }, "Expected to show groups " + expectedListingStable + ", but found " + foundListing[0]);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testConsumerGroupStatesFromString(String quorum) {
+ scala.collection.Set<ConsumerGroupState> result = ConsumerGroupCommand.consumerGroupStatesFromString("Stable");
+ assertEquals(set(Collections.singleton(ConsumerGroupState.STABLE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("Stable, PreparingRebalance");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.STABLE, ConsumerGroupState.PREPARING_REBALANCE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("Dead,CompletingRebalance,");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.DEAD, ConsumerGroupState.COMPLETING_REBALANCE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("stable");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.STABLE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("stable, assigning");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.STABLE, ConsumerGroupState.ASSIGNING)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("dead,reconciling,");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.DEAD, ConsumerGroupState.RECONCILING)), result);
+
+ assertThrows(IllegalArgumentException.class, () -> ConsumerGroupCommand.consumerGroupStatesFromString("bad, wrong"));
+
+ assertThrows(IllegalArgumentException.class, () -> ConsumerGroupCommand.consumerGroupStatesFromString(" bad, Stable"));
+
+ assertThrows(IllegalArgumentException.class, () -> ConsumerGroupCommand.consumerGroupStatesFromString(" , ,"));
Review Comment:
It seems 1 test is missing.
##########
tools/src/test/java/org/apache/kafka/tools/consumer/group/ListConsumerGroupTest.java:
##########
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.tools.consumer.group;
+
+import joptsimple.OptionException;
+import kafka.admin.ConsumerGroupCommand;
+import org.apache.kafka.clients.admin.ConsumerGroupListing;
+import org.apache.kafka.common.ConsumerGroupState;
+import org.apache.kafka.test.TestUtils;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Objects;
+import java.util.Optional;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class ListConsumerGroupTest extends ConsumerGroupCommandTest {
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListConsumerGroups(String quorum) throws Exception {
+ String simpleGroup = "simple-group";
+ addSimpleGroupExecutor(simpleGroup);
+ addConsumerGroupExecutor(1);
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list"};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+ scala.collection.Set<String> expectedGroups = set(Arrays.asList(GROUP, simpleGroup));
+ final scala.collection.Set[] foundGroups = new scala.collection.Set[1];
+ TestUtils.waitForCondition(() -> {
+ foundGroups[0] = service.listConsumerGroups().toSet();
+ return Objects.equals(expectedGroups, foundGroups[0]);
+ }, "Expected --list to show groups " + expectedGroups + ", but found " + foundGroups[0] + ".");
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListWithUnrecognizedNewConsumerOption() {
+ String[] cgcArgs = new String[]{"--new-consumer", "--bootstrap-server", bootstrapServers(listenerName()), "--list"};
+ assertThrows(OptionException.class, () -> getConsumerGroupService(cgcArgs));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListConsumerGroupsWithStates() throws Exception {
+ String simpleGroup = "simple-group";
+ addSimpleGroupExecutor(simpleGroup);
+ addConsumerGroupExecutor(1);
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list", "--state"};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ scala.collection.Set<ConsumerGroupListing> expectedListing = set(Arrays.asList(
+ new ConsumerGroupListing(simpleGroup, true, Optional.of(ConsumerGroupState.EMPTY)),
+ new ConsumerGroupListing(GROUP, false, Optional.of(ConsumerGroupState.STABLE))));
+
+ final scala.collection.Set[] foundListing = new scala.collection.Set[1];
+ TestUtils.waitForCondition(() -> {
+ foundListing[0] = service.listConsumerGroupsWithState(set(Arrays.asList(ConsumerGroupState.values()))).toSet();
+ return Objects.equals(expectedListing, foundListing[0]);
+ }, "Expected to show groups " + expectedListing + ", but found " + foundListing[0]);
+
+ scala.collection.Set<ConsumerGroupListing> expectedListingStable = set(Collections.singleton(
+ new ConsumerGroupListing(GROUP, false, Optional.of(ConsumerGroupState.STABLE))));
+
+ foundListing[0] = null;
+
+ TestUtils.waitForCondition(() -> {
+ foundListing[0] = service.listConsumerGroupsWithState(set(Collections.singleton(ConsumerGroupState.STABLE))).toSet();
+ return Objects.equals(expectedListingStable, foundListing[0]);
+ }, "Expected to show groups " + expectedListingStable + ", but found " + foundListing[0]);
Review Comment:
This is hard to read. `foundListing[0] = null` doesn't make sense to me. Please improve it. Thanks.
##########
tools/src/test/java/org/apache/kafka/tools/consumer/group/DeleteConsumerGroupsTest.java:
##########
@@ -0,0 +1,317 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.tools.consumer.group;
+
+import joptsimple.OptionException;
+import kafka.admin.ConsumerGroupCommand;
+import org.apache.kafka.clients.consumer.RangeAssignor;
+import org.apache.kafka.common.errors.GroupIdNotFoundException;
+import org.apache.kafka.common.errors.GroupNotEmptyException;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.test.TestUtils;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.kafka.tools.ToolsTestUtils.TEST_WITH_PARAMETERIZED_QUORUM_NAME;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class DeleteConsumerGroupsTest extends ConsumerGroupCommandTest {
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteWithTopicOption(String quorum) {
+ createOffsetsTopic(listenerName(), new Properties());
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", GROUP, "--topic"};
+ assertThrows(OptionException.class, () -> getConsumerGroupService(cgcArgs));
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteCmdNonExistingGroup(String quorum) {
+ createOffsetsTopic(listenerName(), new Properties());
+ String missingGroup = "missing.group";
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", missingGroup};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ String output = kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ service.deleteGroups();
+ return null;
+ });
+ assertTrue(output.contains("Group '" + missingGroup + "' could not be deleted due to:") && output.contains(Errors.GROUP_ID_NOT_FOUND.message()),
+ "The expected error (" + Errors.GROUP_ID_NOT_FOUND + ") was not detected while deleting consumer group");
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteNonExistingGroup(String quorum) {
+ createOffsetsTopic(listenerName(), new Properties());
+ String missingGroup = "missing.group";
+
+ // note the group to be deleted is a different (non-existing) group
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", missingGroup};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ scala.collection.Map<String, Throwable> result = service.deleteGroups();
+ assertTrue(result.size() == 1 && result.contains(missingGroup) && result.get(missingGroup).get().getCause() instanceof GroupIdNotFoundException,
+ "The expected error (" + Errors.GROUP_ID_NOT_FOUND + ") was not detected while deleting consumer group");
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteCmdNonEmptyGroup(String quorum) throws Exception {
+ createOffsetsTopic(listenerName(), new Properties());
+
+ // run one consumer in the group
+ addConsumerGroupExecutor(1);
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", GROUP};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ TestUtils.waitForCondition(
+ () -> service.collectGroupMembers(GROUP, false)._2.get().size() == 1,
+ "The group did not initialize as expected."
+ );
+
+ String output = kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ service.deleteGroups();
+ return null;
Review Comment:
Will it have compiling failure after removing `return null`?
##########
tools/src/test/java/org/apache/kafka/tools/consumer/group/ListConsumerGroupTest.java:
##########
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.tools.consumer.group;
+
+import joptsimple.OptionException;
+import kafka.admin.ConsumerGroupCommand;
+import org.apache.kafka.clients.admin.ConsumerGroupListing;
+import org.apache.kafka.common.ConsumerGroupState;
+import org.apache.kafka.test.TestUtils;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Objects;
+import java.util.Optional;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class ListConsumerGroupTest extends ConsumerGroupCommandTest {
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListConsumerGroups(String quorum) throws Exception {
+ String simpleGroup = "simple-group";
+ addSimpleGroupExecutor(simpleGroup);
+ addConsumerGroupExecutor(1);
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list"};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+ scala.collection.Set<String> expectedGroups = set(Arrays.asList(GROUP, simpleGroup));
+ final scala.collection.Set[] foundGroups = new scala.collection.Set[1];
+ TestUtils.waitForCondition(() -> {
+ foundGroups[0] = service.listConsumerGroups().toSet();
+ return Objects.equals(expectedGroups, foundGroups[0]);
+ }, "Expected --list to show groups " + expectedGroups + ", but found " + foundGroups[0] + ".");
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListWithUnrecognizedNewConsumerOption() {
+ String[] cgcArgs = new String[]{"--new-consumer", "--bootstrap-server", bootstrapServers(listenerName()), "--list"};
+ assertThrows(OptionException.class, () -> getConsumerGroupService(cgcArgs));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListConsumerGroupsWithStates() throws Exception {
+ String simpleGroup = "simple-group";
+ addSimpleGroupExecutor(simpleGroup);
+ addConsumerGroupExecutor(1);
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list", "--state"};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ scala.collection.Set<ConsumerGroupListing> expectedListing = set(Arrays.asList(
+ new ConsumerGroupListing(simpleGroup, true, Optional.of(ConsumerGroupState.EMPTY)),
+ new ConsumerGroupListing(GROUP, false, Optional.of(ConsumerGroupState.STABLE))));
+
+ final scala.collection.Set[] foundListing = new scala.collection.Set[1];
+ TestUtils.waitForCondition(() -> {
+ foundListing[0] = service.listConsumerGroupsWithState(set(Arrays.asList(ConsumerGroupState.values()))).toSet();
+ return Objects.equals(expectedListing, foundListing[0]);
+ }, "Expected to show groups " + expectedListing + ", but found " + foundListing[0]);
+
+ scala.collection.Set<ConsumerGroupListing> expectedListingStable = set(Collections.singleton(
+ new ConsumerGroupListing(GROUP, false, Optional.of(ConsumerGroupState.STABLE))));
+
+ foundListing[0] = null;
+
+ TestUtils.waitForCondition(() -> {
+ foundListing[0] = service.listConsumerGroupsWithState(set(Collections.singleton(ConsumerGroupState.STABLE))).toSet();
+ return Objects.equals(expectedListingStable, foundListing[0]);
+ }, "Expected to show groups " + expectedListingStable + ", but found " + foundListing[0]);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testConsumerGroupStatesFromString(String quorum) {
+ scala.collection.Set<ConsumerGroupState> result = ConsumerGroupCommand.consumerGroupStatesFromString("Stable");
+ assertEquals(set(Collections.singleton(ConsumerGroupState.STABLE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("Stable, PreparingRebalance");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.STABLE, ConsumerGroupState.PREPARING_REBALANCE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("Dead,CompletingRebalance,");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.DEAD, ConsumerGroupState.COMPLETING_REBALANCE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("stable");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.STABLE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("stable, assigning");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.STABLE, ConsumerGroupState.ASSIGNING)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("dead,reconciling,");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.DEAD, ConsumerGroupState.RECONCILING)), result);
Review Comment:
I didn't see these tests in scala. Did you add them?
##########
tools/src/test/java/org/apache/kafka/tools/consumer/group/ListConsumerGroupTest.java:
##########
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.tools.consumer.group;
+
+import joptsimple.OptionException;
+import kafka.admin.ConsumerGroupCommand;
+import org.apache.kafka.clients.admin.ConsumerGroupListing;
+import org.apache.kafka.common.ConsumerGroupState;
+import org.apache.kafka.test.TestUtils;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Objects;
+import java.util.Optional;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class ListConsumerGroupTest extends ConsumerGroupCommandTest {
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListConsumerGroups(String quorum) throws Exception {
+ String simpleGroup = "simple-group";
+ addSimpleGroupExecutor(simpleGroup);
+ addConsumerGroupExecutor(1);
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list"};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+ scala.collection.Set<String> expectedGroups = set(Arrays.asList(GROUP, simpleGroup));
+ final scala.collection.Set[] foundGroups = new scala.collection.Set[1];
+ TestUtils.waitForCondition(() -> {
+ foundGroups[0] = service.listConsumerGroups().toSet();
+ return Objects.equals(expectedGroups, foundGroups[0]);
+ }, "Expected --list to show groups " + expectedGroups + ", but found " + foundGroups[0] + ".");
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListWithUnrecognizedNewConsumerOption() {
+ String[] cgcArgs = new String[]{"--new-consumer", "--bootstrap-server", bootstrapServers(listenerName()), "--list"};
+ assertThrows(OptionException.class, () -> getConsumerGroupService(cgcArgs));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListConsumerGroupsWithStates() throws Exception {
+ String simpleGroup = "simple-group";
+ addSimpleGroupExecutor(simpleGroup);
+ addConsumerGroupExecutor(1);
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list", "--state"};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ scala.collection.Set<ConsumerGroupListing> expectedListing = set(Arrays.asList(
+ new ConsumerGroupListing(simpleGroup, true, Optional.of(ConsumerGroupState.EMPTY)),
+ new ConsumerGroupListing(GROUP, false, Optional.of(ConsumerGroupState.STABLE))));
+
+ final scala.collection.Set[] foundListing = new scala.collection.Set[1];
+ TestUtils.waitForCondition(() -> {
+ foundListing[0] = service.listConsumerGroupsWithState(set(Arrays.asList(ConsumerGroupState.values()))).toSet();
+ return Objects.equals(expectedListing, foundListing[0]);
+ }, "Expected to show groups " + expectedListing + ", but found " + foundListing[0]);
+
+ scala.collection.Set<ConsumerGroupListing> expectedListingStable = set(Collections.singleton(
+ new ConsumerGroupListing(GROUP, false, Optional.of(ConsumerGroupState.STABLE))));
+
+ foundListing[0] = null;
+
+ TestUtils.waitForCondition(() -> {
+ foundListing[0] = service.listConsumerGroupsWithState(set(Collections.singleton(ConsumerGroupState.STABLE))).toSet();
+ return Objects.equals(expectedListingStable, foundListing[0]);
+ }, "Expected to show groups " + expectedListingStable + ", but found " + foundListing[0]);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testConsumerGroupStatesFromString(String quorum) {
+ scala.collection.Set<ConsumerGroupState> result = ConsumerGroupCommand.consumerGroupStatesFromString("Stable");
+ assertEquals(set(Collections.singleton(ConsumerGroupState.STABLE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("Stable, PreparingRebalance");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.STABLE, ConsumerGroupState.PREPARING_REBALANCE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("Dead,CompletingRebalance,");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.DEAD, ConsumerGroupState.COMPLETING_REBALANCE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("stable");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.STABLE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("stable, assigning");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.STABLE, ConsumerGroupState.ASSIGNING)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("dead,reconciling,");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.DEAD, ConsumerGroupState.RECONCILING)), result);
+
+ assertThrows(IllegalArgumentException.class, () -> ConsumerGroupCommand.consumerGroupStatesFromString("bad, wrong"));
+
+ assertThrows(IllegalArgumentException.class, () -> ConsumerGroupCommand.consumerGroupStatesFromString(" bad, Stable"));
+
+ assertThrows(IllegalArgumentException.class, () -> ConsumerGroupCommand.consumerGroupStatesFromString(" , ,"));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListGroupCommand(String quorum) throws Exception {
+ String simpleGroup = "simple-group";
+ addSimpleGroupExecutor(simpleGroup);
+ addConsumerGroupExecutor(1);
+ final String[] out = {""};
Review Comment:
Why can't we declare as `String` directly?
`final String out = ""`?
##########
tools/src/test/java/org/apache/kafka/tools/consumer/group/ListConsumerGroupTest.java:
##########
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.tools.consumer.group;
+
+import joptsimple.OptionException;
+import kafka.admin.ConsumerGroupCommand;
+import org.apache.kafka.clients.admin.ConsumerGroupListing;
+import org.apache.kafka.common.ConsumerGroupState;
+import org.apache.kafka.test.TestUtils;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Objects;
+import java.util.Optional;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class ListConsumerGroupTest extends ConsumerGroupCommandTest {
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListConsumerGroups(String quorum) throws Exception {
+ String simpleGroup = "simple-group";
+ addSimpleGroupExecutor(simpleGroup);
+ addConsumerGroupExecutor(1);
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list"};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+ scala.collection.Set<String> expectedGroups = set(Arrays.asList(GROUP, simpleGroup));
+ final scala.collection.Set[] foundGroups = new scala.collection.Set[1];
Review Comment:
Could we use java set? It's weird we compare the 1st element only below. That is, you create a Set, and then, you just verify the 1st element of the Set. Then, when do you need a set if only 1 element?
##########
tools/src/test/java/org/apache/kafka/tools/consumer/group/DeleteConsumerGroupsTest.java:
##########
@@ -0,0 +1,317 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.tools.consumer.group;
+
+import joptsimple.OptionException;
+import kafka.admin.ConsumerGroupCommand;
+import org.apache.kafka.clients.consumer.RangeAssignor;
+import org.apache.kafka.common.errors.GroupIdNotFoundException;
+import org.apache.kafka.common.errors.GroupNotEmptyException;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.test.TestUtils;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.kafka.tools.ToolsTestUtils.TEST_WITH_PARAMETERIZED_QUORUM_NAME;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class DeleteConsumerGroupsTest extends ConsumerGroupCommandTest {
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteWithTopicOption(String quorum) {
+ createOffsetsTopic(listenerName(), new Properties());
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", GROUP, "--topic"};
+ assertThrows(OptionException.class, () -> getConsumerGroupService(cgcArgs));
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteCmdNonExistingGroup(String quorum) {
+ createOffsetsTopic(listenerName(), new Properties());
+ String missingGroup = "missing.group";
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", missingGroup};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ String output = kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ service.deleteGroups();
+ return null;
+ });
+ assertTrue(output.contains("Group '" + missingGroup + "' could not be deleted due to:") && output.contains(Errors.GROUP_ID_NOT_FOUND.message()),
+ "The expected error (" + Errors.GROUP_ID_NOT_FOUND + ") was not detected while deleting consumer group");
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteNonExistingGroup(String quorum) {
+ createOffsetsTopic(listenerName(), new Properties());
+ String missingGroup = "missing.group";
+
+ // note the group to be deleted is a different (non-existing) group
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", missingGroup};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ scala.collection.Map<String, Throwable> result = service.deleteGroups();
+ assertTrue(result.size() == 1 && result.contains(missingGroup) && result.get(missingGroup).get().getCause() instanceof GroupIdNotFoundException,
+ "The expected error (" + Errors.GROUP_ID_NOT_FOUND + ") was not detected while deleting consumer group");
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteCmdNonEmptyGroup(String quorum) throws Exception {
+ createOffsetsTopic(listenerName(), new Properties());
+
+ // run one consumer in the group
+ addConsumerGroupExecutor(1);
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", GROUP};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ TestUtils.waitForCondition(
+ () -> service.collectGroupMembers(GROUP, false)._2.get().size() == 1,
+ "The group did not initialize as expected."
+ );
+
+ String output = kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ service.deleteGroups();
+ return null;
+ });
+ assertTrue(output.contains("Group '" + GROUP + "' could not be deleted due to:") && output.contains(Errors.NON_EMPTY_GROUP.message()),
+ "The expected error (" + Errors.NON_EMPTY_GROUP + ") was not detected while deleting consumer group. Output was: (" + output + ")");
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteNonEmptyGroup(String quorum) throws Exception {
+ createOffsetsTopic(listenerName(), new Properties());
+
+ // run one consumer in the group
+ addConsumerGroupExecutor(1);
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", GROUP};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ TestUtils.waitForCondition(
+ () -> service.collectGroupMembers(GROUP, false)._2.get().size() == 1,
+ "The group did not initialize as expected."
+ );
+
+ scala.collection.Map<String, Throwable> result = service.deleteGroups();
+ assertNotNull(result.get(GROUP).get(),
+ "Group was deleted successfully, but it shouldn't have been. Result was:(" + result + ")");
+ assertTrue(result.size() == 1 && result.contains(GROUP) && result.get(GROUP).get().getCause() instanceof GroupNotEmptyException,
+ "The expected error (" + Errors.NON_EMPTY_GROUP + ") was not detected while deleting consumer group. Result was:(" + result + ")");
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteCmdEmptyGroup(String quorum) throws Exception {
+ createOffsetsTopic(listenerName(), new Properties());
+
+ // run one consumer in the group
+ ConsumerGroupExecutor executor = addConsumerGroupExecutor(1);
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", GROUP};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ TestUtils.waitForCondition(
+ () -> service.listConsumerGroups().contains(GROUP) && Objects.equals(service.collectGroupState(GROUP).state(), "Stable"),
+ "The group did not initialize as expected."
+ );
+
+ executor.shutdown();
+
+ TestUtils.waitForCondition(
+ () -> Objects.equals(service.collectGroupState(GROUP).state(), "Empty"),
+ "The group did not become empty as expected."
+ );
+
+ String output = kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ service.deleteGroups();
+ return null;
+ });
+ assertTrue(output.contains("Deletion of requested consumer groups ('" + GROUP + "') was successful."),
+ "The consumer group could not be deleted as expected");
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteCmdAllGroups(String quorum) throws Exception {
+ createOffsetsTopic(listenerName(), new Properties());
+
+ // Create 3 groups with 1 consumer per each
+ Map<String, ConsumerGroupExecutor> groups = IntStream.rangeClosed(1, 3).mapToObj(i -> GROUP + i).collect(Collectors.toMap(
+ Function.identity(),
+ group -> addConsumerGroupExecutor(1, TOPIC, group, RangeAssignor.class.getName(), Optional.empty(), false)
+ ));
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--all-groups"};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ TestUtils.waitForCondition(() ->
+ Objects.equals(service.listConsumerGroups().toSet(), set(groups.keySet())) &&
+ groups.keySet().stream().allMatch(groupId -> {
+ try {
+ return Objects.equals(service.collectGroupState(groupId).state(), "Stable");
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }),
+ "The group did not initialize as expected.");
+
+ // Shutdown consumers to empty out groups
+ groups.values().forEach(AbstractConsumerGroupExecutor::shutdown);
+
+ TestUtils.waitForCondition(() ->
+ groups.keySet().stream().allMatch(groupId -> {
+ try {
+ return Objects.equals(service.collectGroupState(groupId).state(), "Empty");
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }),
+ "The group did not become empty as expected.");
+
+ String output = kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ service.deleteGroups();
+ return null;
+ }).trim();
+ Set<String> expectedGroupsForDeletion = groups.keySet();
+ Set<String> deletedGroupsGrepped = Arrays.stream(output.substring(output.indexOf('(') + 1, output.indexOf(')')).split(","))
+ .map(str -> str.replaceAll("'", "").trim()).collect(Collectors.toSet());
+
+ assertTrue(output.matches("Deletion of requested consumer groups (.*) was successful.")
+ && Objects.equals(deletedGroupsGrepped, expectedGroupsForDeletion),
+ "The consumer group(s) could not be deleted as expected");
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteEmptyGroup(String quorum) throws Exception {
+ createOffsetsTopic(listenerName(), new Properties());
+
+ // run one consumer in the group
+ ConsumerGroupExecutor executor = addConsumerGroupExecutor(1);
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", GROUP};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ TestUtils.waitForCondition(
+ () -> service.listConsumerGroups().contains(GROUP) && Objects.equals(service.collectGroupState(GROUP).state(), "Stable"),
+ "The group did not initialize as expected.");
+
+ executor.shutdown();
+
+ TestUtils.waitForCondition(
+ () -> Objects.equals(service.collectGroupState(GROUP).state(), "Empty"),
+ "The group did not become empty as expected.");
+
+ scala.collection.Map<String, Throwable> result = service.deleteGroups();
+ assertTrue(result.size() == 1 && result.contains(GROUP) && result.get(GROUP).get() == null,
+ "The consumer group could not be deleted as expected");
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteCmdWithMixOfSuccessAndError(String quorum) throws Exception {
+ createOffsetsTopic(listenerName(), new Properties());
+ String missingGroup = "missing.group";
+
+ // run one consumer in the group
+ ConsumerGroupExecutor executor = addConsumerGroupExecutor(1);
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", GROUP};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ TestUtils.waitForCondition(
+ () -> service.listConsumerGroups().contains(GROUP) && Objects.equals(service.collectGroupState(GROUP).state(), "Stable"),
+ "The group did not initialize as expected.");
+
+ executor.shutdown();
+
+ TestUtils.waitForCondition(
+ () -> Objects.equals(service.collectGroupState(GROUP).state(), "Empty"),
+ "The group did not become empty as expected.");
+
+ String[] cgcArgs2 = new String[cgcArgs.length + 2];
+
+ System.arraycopy(cgcArgs, 0, cgcArgs2, 0, cgcArgs.length);
+ cgcArgs2[cgcArgs2.length - 2] = "--group";
+ cgcArgs2[cgcArgs2.length - 1] = missingGroup;
Review Comment:
I know we did that in scala like this, but after rewriting in Java it becomes more unreadable. Maybe we can just do this:
```
String[] cgcArgs2 = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", GROUP, "--group", missingGroup};
```
##########
tools/src/test/java/org/apache/kafka/tools/consumer/group/ListConsumerGroupTest.java:
##########
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.tools.consumer.group;
+
+import joptsimple.OptionException;
+import kafka.admin.ConsumerGroupCommand;
+import org.apache.kafka.clients.admin.ConsumerGroupListing;
+import org.apache.kafka.common.ConsumerGroupState;
+import org.apache.kafka.test.TestUtils;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Objects;
+import java.util.Optional;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class ListConsumerGroupTest extends ConsumerGroupCommandTest {
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListConsumerGroups(String quorum) throws Exception {
+ String simpleGroup = "simple-group";
+ addSimpleGroupExecutor(simpleGroup);
+ addConsumerGroupExecutor(1);
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list"};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+ scala.collection.Set<String> expectedGroups = set(Arrays.asList(GROUP, simpleGroup));
+ final scala.collection.Set[] foundGroups = new scala.collection.Set[1];
+ TestUtils.waitForCondition(() -> {
+ foundGroups[0] = service.listConsumerGroups().toSet();
+ return Objects.equals(expectedGroups, foundGroups[0]);
+ }, "Expected --list to show groups " + expectedGroups + ", but found " + foundGroups[0] + ".");
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListWithUnrecognizedNewConsumerOption() {
+ String[] cgcArgs = new String[]{"--new-consumer", "--bootstrap-server", bootstrapServers(listenerName()), "--list"};
+ assertThrows(OptionException.class, () -> getConsumerGroupService(cgcArgs));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListConsumerGroupsWithStates() throws Exception {
+ String simpleGroup = "simple-group";
+ addSimpleGroupExecutor(simpleGroup);
+ addConsumerGroupExecutor(1);
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list", "--state"};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ scala.collection.Set<ConsumerGroupListing> expectedListing = set(Arrays.asList(
+ new ConsumerGroupListing(simpleGroup, true, Optional.of(ConsumerGroupState.EMPTY)),
+ new ConsumerGroupListing(GROUP, false, Optional.of(ConsumerGroupState.STABLE))));
+
+ final scala.collection.Set[] foundListing = new scala.collection.Set[1];
+ TestUtils.waitForCondition(() -> {
+ foundListing[0] = service.listConsumerGroupsWithState(set(Arrays.asList(ConsumerGroupState.values()))).toSet();
+ return Objects.equals(expectedListing, foundListing[0]);
+ }, "Expected to show groups " + expectedListing + ", but found " + foundListing[0]);
+
+ scala.collection.Set<ConsumerGroupListing> expectedListingStable = set(Collections.singleton(
+ new ConsumerGroupListing(GROUP, false, Optional.of(ConsumerGroupState.STABLE))));
+
+ foundListing[0] = null;
+
+ TestUtils.waitForCondition(() -> {
+ foundListing[0] = service.listConsumerGroupsWithState(set(Collections.singleton(ConsumerGroupState.STABLE))).toSet();
+ return Objects.equals(expectedListingStable, foundListing[0]);
+ }, "Expected to show groups " + expectedListingStable + ", but found " + foundListing[0]);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testConsumerGroupStatesFromString(String quorum) {
+ scala.collection.Set<ConsumerGroupState> result = ConsumerGroupCommand.consumerGroupStatesFromString("Stable");
+ assertEquals(set(Collections.singleton(ConsumerGroupState.STABLE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("Stable, PreparingRebalance");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.STABLE, ConsumerGroupState.PREPARING_REBALANCE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("Dead,CompletingRebalance,");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.DEAD, ConsumerGroupState.COMPLETING_REBALANCE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("stable");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.STABLE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("stable, assigning");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.STABLE, ConsumerGroupState.ASSIGNING)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("dead,reconciling,");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.DEAD, ConsumerGroupState.RECONCILING)), result);
+
+ assertThrows(IllegalArgumentException.class, () -> ConsumerGroupCommand.consumerGroupStatesFromString("bad, wrong"));
+
+ assertThrows(IllegalArgumentException.class, () -> ConsumerGroupCommand.consumerGroupStatesFromString(" bad, Stable"));
+
+ assertThrows(IllegalArgumentException.class, () -> ConsumerGroupCommand.consumerGroupStatesFromString(" , ,"));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListGroupCommand(String quorum) throws Exception {
+ String simpleGroup = "simple-group";
+ addSimpleGroupExecutor(simpleGroup);
+ addConsumerGroupExecutor(1);
+ final String[] out = {""};
+
+ String[] cgcArgs1 = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list"};
+ TestUtils.waitForCondition(() -> {
+ out[0] = kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ ConsumerGroupCommand.main(cgcArgs1);
+ return null;
+ });
+ return !out[0].contains("STATE") && out[0].contains(simpleGroup) && out[0].contains(GROUP);
+ }, "Expected to find " + simpleGroup + ", " + GROUP + " and no header, but found " + out[0]);
+
+ String[] cgcArgs2 = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list", "--state"};
+ TestUtils.waitForCondition(() -> {
+ out[0] = kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ ConsumerGroupCommand.main(cgcArgs2);
+ return null;
+ });
+ return out[0].contains("STATE") && out[0].contains(simpleGroup) && out[0].contains(GROUP);
+ }, "Expected to find " + simpleGroup + ", " + GROUP + " and the header, but found " + out[0]);
+
+ String[] cgcArgs3 = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list", "--state", "Stable"};
+ TestUtils.waitForCondition(() -> {
+ out[0] = kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ ConsumerGroupCommand.main(cgcArgs3);
+ return null;
+ });
+ return out[0].contains("STATE") && out[0].contains(GROUP) && out[0].contains("Stable");
+ }, "Expected to find " + GROUP + " in state Stable and the header, but found " + out[0]);
+
+ String[] cgcArgs4 = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list", "--state", "stable"};
+ TestUtils.waitForCondition(() -> {
+ out[0] = kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ ConsumerGroupCommand.main(cgcArgs4);
+ return null;
+ });
+ return out[0].contains("STATE") && out[0].contains(GROUP) && out[0].contains("Stable");
+ }, "Expected to find " + GROUP + " in state Stable and the header, but found " + out[0]);
Review Comment:
Duplicate?
##########
tools/src/test/java/org/apache/kafka/tools/consumer/group/DeleteConsumerGroupsTest.java:
##########
@@ -0,0 +1,317 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.tools.consumer.group;
+
+import joptsimple.OptionException;
+import kafka.admin.ConsumerGroupCommand;
+import org.apache.kafka.clients.consumer.RangeAssignor;
+import org.apache.kafka.common.errors.GroupIdNotFoundException;
+import org.apache.kafka.common.errors.GroupNotEmptyException;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.test.TestUtils;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.kafka.tools.ToolsTestUtils.TEST_WITH_PARAMETERIZED_QUORUM_NAME;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class DeleteConsumerGroupsTest extends ConsumerGroupCommandTest {
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteWithTopicOption(String quorum) {
+ createOffsetsTopic(listenerName(), new Properties());
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", GROUP, "--topic"};
+ assertThrows(OptionException.class, () -> getConsumerGroupService(cgcArgs));
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteCmdNonExistingGroup(String quorum) {
+ createOffsetsTopic(listenerName(), new Properties());
+ String missingGroup = "missing.group";
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", missingGroup};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ String output = kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ service.deleteGroups();
+ return null;
+ });
+ assertTrue(output.contains("Group '" + missingGroup + "' could not be deleted due to:") && output.contains(Errors.GROUP_ID_NOT_FOUND.message()),
+ "The expected error (" + Errors.GROUP_ID_NOT_FOUND + ") was not detected while deleting consumer group");
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteNonExistingGroup(String quorum) {
+ createOffsetsTopic(listenerName(), new Properties());
+ String missingGroup = "missing.group";
+
+ // note the group to be deleted is a different (non-existing) group
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", missingGroup};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ scala.collection.Map<String, Throwable> result = service.deleteGroups();
+ assertTrue(result.size() == 1 && result.contains(missingGroup) && result.get(missingGroup).get().getCause() instanceof GroupIdNotFoundException,
+ "The expected error (" + Errors.GROUP_ID_NOT_FOUND + ") was not detected while deleting consumer group");
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteCmdNonEmptyGroup(String quorum) throws Exception {
+ createOffsetsTopic(listenerName(), new Properties());
+
+ // run one consumer in the group
+ addConsumerGroupExecutor(1);
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", GROUP};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ TestUtils.waitForCondition(
+ () -> service.collectGroupMembers(GROUP, false)._2.get().size() == 1,
+ "The group did not initialize as expected."
+ );
+
+ String output = kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ service.deleteGroups();
+ return null;
+ });
+ assertTrue(output.contains("Group '" + GROUP + "' could not be deleted due to:") && output.contains(Errors.NON_EMPTY_GROUP.message()),
+ "The expected error (" + Errors.NON_EMPTY_GROUP + ") was not detected while deleting consumer group. Output was: (" + output + ")");
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteNonEmptyGroup(String quorum) throws Exception {
+ createOffsetsTopic(listenerName(), new Properties());
+
+ // run one consumer in the group
+ addConsumerGroupExecutor(1);
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", GROUP};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ TestUtils.waitForCondition(
+ () -> service.collectGroupMembers(GROUP, false)._2.get().size() == 1,
+ "The group did not initialize as expected."
+ );
+
+ scala.collection.Map<String, Throwable> result = service.deleteGroups();
+ assertNotNull(result.get(GROUP).get(),
+ "Group was deleted successfully, but it shouldn't have been. Result was:(" + result + ")");
+ assertTrue(result.size() == 1 && result.contains(GROUP) && result.get(GROUP).get().getCause() instanceof GroupNotEmptyException,
+ "The expected error (" + Errors.NON_EMPTY_GROUP + ") was not detected while deleting consumer group. Result was:(" + result + ")");
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteCmdEmptyGroup(String quorum) throws Exception {
+ createOffsetsTopic(listenerName(), new Properties());
+
+ // run one consumer in the group
+ ConsumerGroupExecutor executor = addConsumerGroupExecutor(1);
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", GROUP};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ TestUtils.waitForCondition(
+ () -> service.listConsumerGroups().contains(GROUP) && Objects.equals(service.collectGroupState(GROUP).state(), "Stable"),
+ "The group did not initialize as expected."
+ );
+
+ executor.shutdown();
+
+ TestUtils.waitForCondition(
+ () -> Objects.equals(service.collectGroupState(GROUP).state(), "Empty"),
+ "The group did not become empty as expected."
+ );
+
+ String output = kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ service.deleteGroups();
+ return null;
+ });
+ assertTrue(output.contains("Deletion of requested consumer groups ('" + GROUP + "') was successful."),
+ "The consumer group could not be deleted as expected");
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteCmdAllGroups(String quorum) throws Exception {
+ createOffsetsTopic(listenerName(), new Properties());
+
+ // Create 3 groups with 1 consumer per each
+ Map<String, ConsumerGroupExecutor> groups = IntStream.rangeClosed(1, 3).mapToObj(i -> GROUP + i).collect(Collectors.toMap(
+ Function.identity(),
+ group -> addConsumerGroupExecutor(1, TOPIC, group, RangeAssignor.class.getName(), Optional.empty(), false)
+ ));
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--all-groups"};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ TestUtils.waitForCondition(() ->
+ Objects.equals(service.listConsumerGroups().toSet(), set(groups.keySet())) &&
+ groups.keySet().stream().allMatch(groupId -> {
+ try {
+ return Objects.equals(service.collectGroupState(groupId).state(), "Stable");
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }),
+ "The group did not initialize as expected.");
+
+ // Shutdown consumers to empty out groups
+ groups.values().forEach(AbstractConsumerGroupExecutor::shutdown);
+
+ TestUtils.waitForCondition(() ->
+ groups.keySet().stream().allMatch(groupId -> {
+ try {
+ return Objects.equals(service.collectGroupState(groupId).state(), "Empty");
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }),
+ "The group did not become empty as expected.");
+
+ String output = kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ service.deleteGroups();
+ return null;
+ }).trim();
+ Set<String> expectedGroupsForDeletion = groups.keySet();
+ Set<String> deletedGroupsGrepped = Arrays.stream(output.substring(output.indexOf('(') + 1, output.indexOf(')')).split(","))
+ .map(str -> str.replaceAll("'", "").trim()).collect(Collectors.toSet());
+
+ assertTrue(output.matches("Deletion of requested consumer groups (.*) was successful.")
+ && Objects.equals(deletedGroupsGrepped, expectedGroupsForDeletion),
+ "The consumer group(s) could not be deleted as expected");
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteEmptyGroup(String quorum) throws Exception {
+ createOffsetsTopic(listenerName(), new Properties());
+
+ // run one consumer in the group
+ ConsumerGroupExecutor executor = addConsumerGroupExecutor(1);
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", GROUP};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ TestUtils.waitForCondition(
+ () -> service.listConsumerGroups().contains(GROUP) && Objects.equals(service.collectGroupState(GROUP).state(), "Stable"),
+ "The group did not initialize as expected.");
+
+ executor.shutdown();
+
+ TestUtils.waitForCondition(
+ () -> Objects.equals(service.collectGroupState(GROUP).state(), "Empty"),
+ "The group did not become empty as expected.");
+
+ scala.collection.Map<String, Throwable> result = service.deleteGroups();
+ assertTrue(result.size() == 1 && result.contains(GROUP) && result.get(GROUP).get() == null,
+ "The consumer group could not be deleted as expected");
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteCmdWithMixOfSuccessAndError(String quorum) throws Exception {
+ createOffsetsTopic(listenerName(), new Properties());
+ String missingGroup = "missing.group";
+
+ // run one consumer in the group
+ ConsumerGroupExecutor executor = addConsumerGroupExecutor(1);
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", GROUP};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ TestUtils.waitForCondition(
+ () -> service.listConsumerGroups().contains(GROUP) && Objects.equals(service.collectGroupState(GROUP).state(), "Stable"),
+ "The group did not initialize as expected.");
+
+ executor.shutdown();
+
+ TestUtils.waitForCondition(
+ () -> Objects.equals(service.collectGroupState(GROUP).state(), "Empty"),
+ "The group did not become empty as expected.");
+
+ String[] cgcArgs2 = new String[cgcArgs.length + 2];
+
+ System.arraycopy(cgcArgs, 0, cgcArgs2, 0, cgcArgs.length);
+ cgcArgs2[cgcArgs2.length - 2] = "--group";
+ cgcArgs2[cgcArgs2.length - 1] = missingGroup;
+
+ ConsumerGroupCommand.ConsumerGroupService service2 = getConsumerGroupService(cgcArgs2);
+
+ String output = kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ service2.deleteGroups();
+ return null;
+ });
+ assertTrue(output.contains("Group '" + missingGroup + "' could not be deleted due to:")
+ && output.contains(Errors.GROUP_ID_NOT_FOUND.message())
+ && output.contains("These consumer groups were deleted successfully: '" + GROUP + "'"),
+ "The consumer group deletion did not work as expected");
+ }
+
+ @ParameterizedTest(name = TEST_WITH_PARAMETERIZED_QUORUM_NAME)
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testDeleteWithMixOfSuccessAndError(String quorum) throws Exception {
+ createOffsetsTopic(listenerName(), new Properties());
+ String missingGroup = "missing.group";
+
+ // run one consumer in the group
+ ConsumerGroupExecutor executor = addConsumerGroupExecutor(1);
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", GROUP};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ TestUtils.waitForCondition(
+ () -> service.listConsumerGroups().contains(GROUP) && Objects.equals(service.collectGroupState(GROUP).state(), "Stable"),
+ "The group did not initialize as expected.");
+
+ executor.shutdown();
+
+ TestUtils.waitForCondition(
+ () -> Objects.equals(service.collectGroupState(GROUP).state(), "Empty"),
+ "The group did not become empty as expected.");
+
+ String[] cgcArgs2 = new String[cgcArgs.length + 2];
+
+ System.arraycopy(cgcArgs, 0, cgcArgs2, 0, cgcArgs.length);
+ cgcArgs2[cgcArgs2.length - 2] = "--group";
+ cgcArgs2[cgcArgs2.length - 1] = missingGroup;
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14589 [1/3] Tests of ConsoleGroupCommand rewritten in java [kafka]
Posted by "nizhikov (via GitHub)" <gi...@apache.org>.
nizhikov commented on PR #15256:
URL: https://github.com/apache/kafka/pull/15256#issuecomment-1940499322
@showuon Thank you for review and merging!
--
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-14589 [1/3] Tests of ConsoleGroupCommand rewritten in java [kafka]
Posted by "nizhikov (via GitHub)" <gi...@apache.org>.
nizhikov commented on PR #15256:
URL: https://github.com/apache/kafka/pull/15256#issuecomment-1939084618
Hello, @mimaison @jolshan
Can you, please, take a look.
Looks like this PR are ready to merge.
--
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-14589 [1/3] Tests of ConsoleGroupCommand rewritten in java [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on PR #15256:
URL: https://github.com/apache/kafka/pull/15256#issuecomment-1940345746
> Hello @showuon
>
> I found that `DeleteTopicTest.scala` don't close resources allocated in case of kraft run. It was hided by removed tests and lead to the tests failures we observed earlier.
>
> I fixed `DeleteTopicTest` and now it seems that CI is OK.
>
> Take a look one more time, please.
Nice find! Thanks for the fix!
--
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-14589 [1/3] Tests of ConsoleGroupCommand rewritten in java [kafka]
Posted by "nizhikov (via GitHub)" <gi...@apache.org>.
nizhikov commented on PR #15256:
URL: https://github.com/apache/kafka/pull/15256#issuecomment-1921873375
Other trunk test run contains `DynamicBrokerReconfigurationTest ` fails, also (but only for 1 environment).
https://ci-builds.apache.org/blue/organizations/jenkins/Kafka%2Fkafka/detail/trunk/2610/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-14589 [1/3] Tests of ConsoleGroupCommand rewritten in java [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on PR #15256:
URL: https://github.com/apache/kafka/pull/15256#issuecomment-1920966427
@nizhikov I merged https://github.com/apache/kafka/pull/15253. I am not sure if it has any impact on this PR. Does it?
--
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-14589 [1/3] Tests of ConsoleGroupCommand rewritten in java [kafka]
Posted by "nizhikov (via GitHub)" <gi...@apache.org>.
nizhikov commented on code in PR #15256:
URL: https://github.com/apache/kafka/pull/15256#discussion_r1471221535
##########
tools/src/test/java/org/apache/kafka/tools/consumer/group/ListConsumerGroupTest.java:
##########
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.tools.consumer.group;
+
+import joptsimple.OptionException;
+import kafka.admin.ConsumerGroupCommand;
+import org.apache.kafka.clients.admin.ConsumerGroupListing;
+import org.apache.kafka.common.ConsumerGroupState;
+import org.apache.kafka.test.TestUtils;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Objects;
+import java.util.Optional;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class ListConsumerGroupTest extends ConsumerGroupCommandTest {
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListConsumerGroups(String quorum) throws Exception {
+ String simpleGroup = "simple-group";
+ addSimpleGroupExecutor(simpleGroup);
+ addConsumerGroupExecutor(1);
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list"};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+ scala.collection.Set<String> expectedGroups = set(Arrays.asList(GROUP, simpleGroup));
+ final scala.collection.Set[] foundGroups = new scala.collection.Set[1];
+ TestUtils.waitForCondition(() -> {
+ foundGroups[0] = service.listConsumerGroups().toSet();
+ return Objects.equals(expectedGroups, foundGroups[0]);
+ }, "Expected --list to show groups " + expectedGroups + ", but found " + foundGroups[0] + ".");
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListWithUnrecognizedNewConsumerOption() {
+ String[] cgcArgs = new String[]{"--new-consumer", "--bootstrap-server", bootstrapServers(listenerName()), "--list"};
+ assertThrows(OptionException.class, () -> getConsumerGroupService(cgcArgs));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListConsumerGroupsWithStates() throws Exception {
+ String simpleGroup = "simple-group";
+ addSimpleGroupExecutor(simpleGroup);
+ addConsumerGroupExecutor(1);
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list", "--state"};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ scala.collection.Set<ConsumerGroupListing> expectedListing = set(Arrays.asList(
+ new ConsumerGroupListing(simpleGroup, true, Optional.of(ConsumerGroupState.EMPTY)),
+ new ConsumerGroupListing(GROUP, false, Optional.of(ConsumerGroupState.STABLE))));
+
+ final scala.collection.Set[] foundListing = new scala.collection.Set[1];
Review Comment:
It seems that usage of scala collections is more convinient for now.
But, of course, when java version of command will be merged we will use java collections here and everywhere in test. You can see how this don for already mereged tests:
https://github.com/apache/kafka/pull/14471/files#diff-4e22e01d567ea1b5087aa60354fa127bb25b892b430a6b86ab39a10edd393c0a
--
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-14589 [1/3] Tests of ConsoleGroupCommand rewritten in java [kafka]
Posted by "nizhikov (via GitHub)" <gi...@apache.org>.
nizhikov commented on code in PR #15256:
URL: https://github.com/apache/kafka/pull/15256#discussion_r1471430390
##########
tools/src/test/java/org/apache/kafka/tools/consumer/group/ListConsumerGroupTest.java:
##########
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.tools.consumer.group;
+
+import joptsimple.OptionException;
+import kafka.admin.ConsumerGroupCommand;
+import org.apache.kafka.clients.admin.ConsumerGroupListing;
+import org.apache.kafka.common.ConsumerGroupState;
+import org.apache.kafka.test.TestUtils;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Objects;
+import java.util.Optional;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class ListConsumerGroupTest extends ConsumerGroupCommandTest {
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListConsumerGroups(String quorum) throws Exception {
+ String simpleGroup = "simple-group";
+ addSimpleGroupExecutor(simpleGroup);
+ addConsumerGroupExecutor(1);
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list"};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+ scala.collection.Set<String> expectedGroups = set(Arrays.asList(GROUP, simpleGroup));
+ final scala.collection.Set[] foundGroups = new scala.collection.Set[1];
+ TestUtils.waitForCondition(() -> {
+ foundGroups[0] = service.listConsumerGroups().toSet();
+ return Objects.equals(expectedGroups, foundGroups[0]);
+ }, "Expected --list to show groups " + expectedGroups + ", but found " + foundGroups[0] + ".");
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListWithUnrecognizedNewConsumerOption() {
+ String[] cgcArgs = new String[]{"--new-consumer", "--bootstrap-server", bootstrapServers(listenerName()), "--list"};
+ assertThrows(OptionException.class, () -> getConsumerGroupService(cgcArgs));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListConsumerGroupsWithStates() throws Exception {
+ String simpleGroup = "simple-group";
+ addSimpleGroupExecutor(simpleGroup);
+ addConsumerGroupExecutor(1);
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list", "--state"};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ scala.collection.Set<ConsumerGroupListing> expectedListing = set(Arrays.asList(
+ new ConsumerGroupListing(simpleGroup, true, Optional.of(ConsumerGroupState.EMPTY)),
+ new ConsumerGroupListing(GROUP, false, Optional.of(ConsumerGroupState.STABLE))));
+
+ final scala.collection.Set[] foundListing = new scala.collection.Set[1];
+ TestUtils.waitForCondition(() -> {
+ foundListing[0] = service.listConsumerGroupsWithState(set(Arrays.asList(ConsumerGroupState.values()))).toSet();
+ return Objects.equals(expectedListing, foundListing[0]);
+ }, "Expected to show groups " + expectedListing + ", but found " + foundListing[0]);
+
+ scala.collection.Set<ConsumerGroupListing> expectedListingStable = set(Collections.singleton(
+ new ConsumerGroupListing(GROUP, false, Optional.of(ConsumerGroupState.STABLE))));
+
+ foundListing[0] = null;
+
+ TestUtils.waitForCondition(() -> {
+ foundListing[0] = service.listConsumerGroupsWithState(set(Collections.singleton(ConsumerGroupState.STABLE))).toSet();
+ return Objects.equals(expectedListingStable, foundListing[0]);
+ }, "Expected to show groups " + expectedListingStable + ", but found " + foundListing[0]);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testConsumerGroupStatesFromString(String quorum) {
+ scala.collection.Set<ConsumerGroupState> result = ConsumerGroupCommand.consumerGroupStatesFromString("Stable");
+ assertEquals(set(Collections.singleton(ConsumerGroupState.STABLE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("Stable, PreparingRebalance");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.STABLE, ConsumerGroupState.PREPARING_REBALANCE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("Dead,CompletingRebalance,");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.DEAD, ConsumerGroupState.COMPLETING_REBALANCE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("stable");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.STABLE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("stable, assigning");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.STABLE, ConsumerGroupState.ASSIGNING)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("dead,reconciling,");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.DEAD, ConsumerGroupState.RECONCILING)), result);
+
+ assertThrows(IllegalArgumentException.class, () -> ConsumerGroupCommand.consumerGroupStatesFromString("bad, wrong"));
+
+ assertThrows(IllegalArgumentException.class, () -> ConsumerGroupCommand.consumerGroupStatesFromString(" bad, Stable"));
+
+ assertThrows(IllegalArgumentException.class, () -> ConsumerGroupCommand.consumerGroupStatesFromString(" , ,"));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListGroupCommand(String quorum) throws Exception {
+ String simpleGroup = "simple-group";
+ addSimpleGroupExecutor(simpleGroup);
+ addConsumerGroupExecutor(1);
+ final String[] out = {""};
+
+ String[] cgcArgs1 = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list"};
+ TestUtils.waitForCondition(() -> {
+ out[0] = kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ ConsumerGroupCommand.main(cgcArgs1);
+ return null;
+ });
+ return !out[0].contains("STATE") && out[0].contains(simpleGroup) && out[0].contains(GROUP);
+ }, "Expected to find " + simpleGroup + ", " + GROUP + " and no header, but found " + out[0]);
+
+ String[] cgcArgs2 = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list", "--state"};
+ TestUtils.waitForCondition(() -> {
+ out[0] = kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ ConsumerGroupCommand.main(cgcArgs2);
+ return null;
+ });
+ return out[0].contains("STATE") && out[0].contains(simpleGroup) && out[0].contains(GROUP);
+ }, "Expected to find " + simpleGroup + ", " + GROUP + " and the header, but found " + out[0]);
+
+ String[] cgcArgs3 = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list", "--state", "Stable"};
+ TestUtils.waitForCondition(() -> {
+ out[0] = kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ ConsumerGroupCommand.main(cgcArgs3);
+ return null;
+ });
+ return out[0].contains("STATE") && out[0].contains(GROUP) && out[0].contains("Stable");
+ }, "Expected to find " + GROUP + " in state Stable and the header, but found " + out[0]);
+
+ String[] cgcArgs4 = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list", "--state", "stable"};
+ TestUtils.waitForCondition(() -> {
+ out[0] = kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ ConsumerGroupCommand.main(cgcArgs4);
+ return null;
+ });
+ return out[0].contains("STATE") && out[0].contains(GROUP) && out[0].contains("Stable");
+ }, "Expected to find " + GROUP + " in state Stable and the header, but found " + out[0]);
Review Comment:
No, case of parameter differs - "Stable", "stable".
--
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-14589 [1/3] Tests of ConsoleGroupCommand rewritten in java [kafka]
Posted by "nizhikov (via GitHub)" <gi...@apache.org>.
nizhikov commented on PR #15256:
URL: https://github.com/apache/kafka/pull/15256#issuecomment-1916886726
@showuon Thanks for review. Answered to all your comments. Please, take a look one more time.
--
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-14589 [1/3] Tests of ConsoleGroupCommand rewritten in java [kafka]
Posted by "nizhikov (via GitHub)" <gi...@apache.org>.
nizhikov commented on code in PR #15256:
URL: https://github.com/apache/kafka/pull/15256#discussion_r1471235686
##########
tools/src/test/java/org/apache/kafka/tools/consumer/group/ListConsumerGroupTest.java:
##########
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.tools.consumer.group;
+
+import joptsimple.OptionException;
+import kafka.admin.ConsumerGroupCommand;
+import org.apache.kafka.clients.admin.ConsumerGroupListing;
+import org.apache.kafka.common.ConsumerGroupState;
+import org.apache.kafka.test.TestUtils;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Objects;
+import java.util.Optional;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class ListConsumerGroupTest extends ConsumerGroupCommandTest {
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListConsumerGroups(String quorum) throws Exception {
+ String simpleGroup = "simple-group";
+ addSimpleGroupExecutor(simpleGroup);
+ addConsumerGroupExecutor(1);
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list"};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+ scala.collection.Set<String> expectedGroups = set(Arrays.asList(GROUP, simpleGroup));
+ final scala.collection.Set[] foundGroups = new scala.collection.Set[1];
+ TestUtils.waitForCondition(() -> {
+ foundGroups[0] = service.listConsumerGroups().toSet();
+ return Objects.equals(expectedGroups, foundGroups[0]);
+ }, "Expected --list to show groups " + expectedGroups + ", but found " + foundGroups[0] + ".");
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListWithUnrecognizedNewConsumerOption() {
+ String[] cgcArgs = new String[]{"--new-consumer", "--bootstrap-server", bootstrapServers(listenerName()), "--list"};
+ assertThrows(OptionException.class, () -> getConsumerGroupService(cgcArgs));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testListConsumerGroupsWithStates() throws Exception {
+ String simpleGroup = "simple-group";
+ addSimpleGroupExecutor(simpleGroup);
+ addConsumerGroupExecutor(1);
+
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list", "--state"};
+ ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs);
+
+ scala.collection.Set<ConsumerGroupListing> expectedListing = set(Arrays.asList(
+ new ConsumerGroupListing(simpleGroup, true, Optional.of(ConsumerGroupState.EMPTY)),
+ new ConsumerGroupListing(GROUP, false, Optional.of(ConsumerGroupState.STABLE))));
+
+ final scala.collection.Set[] foundListing = new scala.collection.Set[1];
+ TestUtils.waitForCondition(() -> {
+ foundListing[0] = service.listConsumerGroupsWithState(set(Arrays.asList(ConsumerGroupState.values()))).toSet();
+ return Objects.equals(expectedListing, foundListing[0]);
+ }, "Expected to show groups " + expectedListing + ", but found " + foundListing[0]);
+
+ scala.collection.Set<ConsumerGroupListing> expectedListingStable = set(Collections.singleton(
+ new ConsumerGroupListing(GROUP, false, Optional.of(ConsumerGroupState.STABLE))));
+
+ foundListing[0] = null;
+
+ TestUtils.waitForCondition(() -> {
+ foundListing[0] = service.listConsumerGroupsWithState(set(Collections.singleton(ConsumerGroupState.STABLE))).toSet();
+ return Objects.equals(expectedListingStable, foundListing[0]);
+ }, "Expected to show groups " + expectedListingStable + ", but found " + foundListing[0]);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"zk", "kraft"})
+ public void testConsumerGroupStatesFromString(String quorum) {
+ scala.collection.Set<ConsumerGroupState> result = ConsumerGroupCommand.consumerGroupStatesFromString("Stable");
+ assertEquals(set(Collections.singleton(ConsumerGroupState.STABLE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("Stable, PreparingRebalance");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.STABLE, ConsumerGroupState.PREPARING_REBALANCE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("Dead,CompletingRebalance,");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.DEAD, ConsumerGroupState.COMPLETING_REBALANCE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("stable");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.STABLE)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("stable, assigning");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.STABLE, ConsumerGroupState.ASSIGNING)), result);
+
+ result = ConsumerGroupCommand.consumerGroupStatesFromString("dead,reconciling,");
+ assertEquals(set(Arrays.asList(ConsumerGroupState.DEAD, ConsumerGroupState.RECONCILING)), result);
Review Comment:
It is fresh changes from 82920ffad00285c11453e64e220ac0ec2666ac54
--
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-14589 [1/3] Tests of ConsoleGroupCommand rewritten in java [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on PR #15256:
URL: https://github.com/apache/kafka/pull/15256#issuecomment-1920760708
@nizhikov , thanks for the reminder, I forgot about this PR.
I just checked https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-15256/8 and https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-15256/7, the test: `kafka.server.DynamicBrokerReconfigurationTest.testThreadPoolResize()` failed on each job once. But it didn't appear in earlier run, https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-15256/4/, as well as the Kafka trunk build: https://ci-builds.apache.org/job/Kafka/job/kafka/job/trunk/2607/ . Could you help check that? Thanks.
--
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-14589 [1/3] Tests of ConsoleGroupCommand rewritten in java [kafka]
Posted by "nizhikov (via GitHub)" <gi...@apache.org>.
nizhikov commented on PR #15256:
URL: https://github.com/apache/kafka/pull/15256#issuecomment-1923260434
Hello @showuon
What do you think? `DynamicBrokerReconfigurationTest` failures related to this PR? Or it can be merged as is?
--
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-14589 [1/3] Tests of ConsoleGroupCommand rewritten in java [kafka]
Posted by "nizhikov (via GitHub)" <gi...@apache.org>.
nizhikov commented on PR #15256:
URL: https://github.com/apache/kafka/pull/15256#issuecomment-1920989673
@showuon I run `DynamicBrokerReconfigurationTest` locally and it passed.
Not sure, how those failures can relate to this PR as I change only specific tests classes.
--
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-14589 [1/3] Tests of ConsoleGroupCommand rewritten in java [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on PR #15256:
URL: https://github.com/apache/kafka/pull/15256#issuecomment-1921076275
@nizhikov , could you rebase with the latest trunk branch and let's see if it fixed the test?
> @showuon I run `DynamicBrokerReconfigurationTest` locally and it passed. Not sure, how those failures can relate to this PR as I change only specific tests classes.
OK, let's wait and see if the latest CI build fixes the test.
--
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