chia7712 commented on code in PR #15766:
URL: https://github.com/apache/kafka/pull/15766#discussion_r1575079922
##########
tools/src/test/java/org/apache/kafka/tools/consumer/group/DeleteConsumerGroupsTest.java:
##########
@@ -17,279 +17,448 @@
package org.apache.kafka.tools.consumer.group;
import joptsimple.OptionException;
+import kafka.test.ClusterInstance;
+import kafka.test.annotation.ClusterConfigProperty;
+import kafka.test.annotation.ClusterTest;
+import kafka.test.annotation.ClusterTestDefaults;
+import kafka.test.annotation.Type;
+import kafka.test.junit.ClusterTestExtensions;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.GroupProtocol;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
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.errors.WakeupException;
import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.test.TestUtils;
import org.apache.kafka.tools.ToolsTestUtils;
-import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.ValueSource;
+import org.junit.jupiter.api.extension.ExtendWith;
+import java.time.Duration;
+import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
import java.util.HashSet;
+import java.util.List;
+import java.util.Locale;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Properties;
import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
import java.util.function.Function;
+import java.util.function.Predicate;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
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
- @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"};
+
+@ExtendWith(value = ClusterTestExtensions.class)
+@ClusterTestDefaults(clusterType = Type.ALL, serverProperties = {
+ @ClusterConfigProperty(key = "offsets.topic.num.partitions", value =
"1"),
+ @ClusterConfigProperty(key = "offsets.topic.replication.factor", value
= "1"),
+})
+public class DeleteConsumerGroupsTest {
+ private final ClusterInstance cluster;
+ private static final String TOPIC = "foo";
+ private static final String GROUP = "test.group";
+
+ public DeleteConsumerGroupsTest(ClusterInstance cluster) {
+ this.cluster = cluster;
+ }
+
+ @ClusterTest
+ public void testDeleteWithTopicOption() {
+ String[] cgcArgs = new String[]{"--bootstrap-server",
cluster.bootstrapServers(), "--delete", "--group", GROUP, "--topic"};
assertThrows(OptionException.class, () ->
getConsumerGroupService(cgcArgs));
}
- @ParameterizedTest
- @ValueSource(strings = {"zk", "kraft"})
- public void testDeleteCmdNonExistingGroup(String quorum) {
- createOffsetsTopic(listenerName(), new Properties());
+ @ClusterTest
+ public void testDeleteCmdNonExistingGroup() {
String missingGroup = "missing.group";
-
- String[] cgcArgs = new String[]{"--bootstrap-server",
bootstrapServers(listenerName()), "--delete", "--group", missingGroup};
+ String[] cgcArgs = new String[]{"--bootstrap-server",
cluster.bootstrapServers(), "--delete", "--group", missingGroup};
ConsumerGroupCommand.ConsumerGroupService service =
getConsumerGroupService(cgcArgs);
-
String output =
ToolsTestUtils.grabConsoleOutput(service::deleteGroups);
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");
+ "The expected error (" + Errors.GROUP_ID_NOT_FOUND + ") was
not detected while deleting consumer group");
}
- @ParameterizedTest
- @ValueSource(strings = {"zk", "kraft"})
- public void testDeleteNonExistingGroup(String quorum) {
- createOffsetsTopic(listenerName(), new Properties());
+ @ClusterTest
+ public void testDeleteNonExistingGroup() {
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};
+ String[] cgcArgs = new String[]{"--bootstrap-server",
cluster.bootstrapServers(), "--delete", "--group", missingGroup};
ConsumerGroupCommand.ConsumerGroupService service =
getConsumerGroupService(cgcArgs);
-
Map<String, Throwable> result = service.deleteGroups();
assertTrue(result.size() == 1 && result.containsKey(missingGroup) &&
result.get(missingGroup).getCause() instanceof GroupIdNotFoundException,
- "The expected error (" + Errors.GROUP_ID_NOT_FOUND + ") was not
detected while deleting consumer group");
+ "The expected error (" + Errors.GROUP_ID_NOT_FOUND + ") was
not detected while deleting consumer group");
}
- @ParameterizedTest
- @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).getValue().get().size() == 1,
- "The group did not initialize as expected."
- );
-
- String output =
ToolsTestUtils.grabConsoleOutput(service::deleteGroups);
- 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 + ")");
+ @ClusterTest
+ public void testDeleteCmdNonEmptyGroup() throws Exception {
+ try (ConsumerGroupExecutor consumerGroupExecutor =
buildConsumerGroupExecutor(GROUP)) {
+ String[] cgcArgs = new String[]{"--bootstrap-server",
cluster.bootstrapServers(), "--delete", "--group", GROUP};
+ ConsumerGroupCommand.ConsumerGroupService service =
getConsumerGroupService(cgcArgs);
+ TestUtils.waitForCondition(
+ () -> service.collectGroupMembers(GROUP,
false).getValue().get().size() == 1,
+ "The group did not initialize as expected."
+ );
+
+ String output =
ToolsTestUtils.grabConsoleOutput(service::deleteGroups);
+ 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
- @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).getValue().get().size() == 1,
- "The group did not initialize as expected."
- );
-
- Map<String, Throwable> result = service.deleteGroups();
- assertNotNull(result.get(GROUP),
- "Group was deleted successfully, but it shouldn't have been.
Result was:(" + result + ")");
- assertTrue(result.size() == 1 && result.containsKey(GROUP) &&
result.get(GROUP).getCause() instanceof GroupNotEmptyException,
- "The expected error (" + Errors.NON_EMPTY_GROUP + ") was not
detected while deleting consumer group. Result was:(" + result + ")");
+ @ClusterTest
+ public void testDeleteNonEmptyGroup() throws Exception {
+ try (ConsumerGroupExecutor consumerGroupExecutor =
buildConsumerGroupExecutor(GROUP)) {
+ String[] cgcArgs = new String[]{"--bootstrap-server",
cluster.bootstrapServers(), "--delete", "--group", GROUP};
+ ConsumerGroupCommand.ConsumerGroupService service =
getConsumerGroupService(cgcArgs);
+
+ TestUtils.waitForCondition(
+ () -> service.collectGroupMembers(GROUP,
false).getValue().get().size() == 1,
+ "The group did not initialize as expected."
+ );
+
+ Map<String, Throwable> result = service.deleteGroups();
+ assertNotNull(result.get(GROUP),
+ "Group was deleted successfully, but it shouldn't have
been. Result was:(" + result + ")");
+ assertTrue(result.size() == 1 && result.containsKey(GROUP) &&
result.get(GROUP).getCause() instanceof GroupNotEmptyException,
+ "The expected error (" + Errors.NON_EMPTY_GROUP + ") was
not detected while deleting consumer group. Result was:(" + result + ")");
+ }
}
- @ParameterizedTest
- @ValueSource(strings = {"zk", "kraft"})
- public void testDeleteCmdEmptyGroup(String quorum) throws Exception {
- createOffsetsTopic(listenerName(), new Properties());
+ @ClusterTest
+ public void testDeleteCmdEmptyGroup() throws Exception {
+ try (ConsumerGroupExecutor consumerGroupExecutor =
buildConsumerGroupExecutor(GROUP)) {
+ String[] cgcArgs = new String[]{"--bootstrap-server",
cluster.bootstrapServers(), "--delete", "--group", GROUP};
+ ConsumerGroupCommand.ConsumerGroupService service =
getConsumerGroupService(cgcArgs);
- // 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."
- );
+ TestUtils.waitForCondition(
+ () -> service.listConsumerGroups().contains(GROUP) &&
Objects.equals(service.collectGroupState(GROUP).state, "Stable"),
+ "The group did not initialize as expected."
+ );
- executor.shutdown();
+ consumerGroupExecutor.close();
- TestUtils.waitForCondition(
- () -> Objects.equals(service.collectGroupState(GROUP).state,
"Empty"),
- "The group did not become empty as expected."
- );
+ TestUtils.waitForCondition(
+ () ->
Objects.equals(service.collectGroupState(GROUP).state, "Empty"),
+ "The group did not become empty as expected."
+ );
- String output =
ToolsTestUtils.grabConsoleOutput(service::deleteGroups);
- assertTrue(output.contains("Deletion of requested consumer groups ('"
+ GROUP + "') was successful."),
- "The consumer group could not be deleted as expected");
+ String output =
ToolsTestUtils.grabConsoleOutput(service::deleteGroups);
+ assertTrue(output.contains("Deletion of requested consumer groups
('" + GROUP + "') was successful."),
+ "The consumer group could not be deleted as expected");
+ }
}
- @ParameterizedTest
- @ValueSource(strings = {"zk", "kraft"})
- public void testDeleteCmdAllGroups(String quorum) throws Exception {
- createOffsetsTopic(listenerName(), new Properties());
-
+ @ClusterTest
+ public void testDeleteCmdAllGroups() throws Exception {
// 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(), Optional.empty(), false,
GroupProtocol.CLASSIC.name)
- ));
+ Map<String, ConsumerGroupExecutor> groupNameToExecutor =
IntStream.rangeClosed(1, 3)
+ .mapToObj(i -> GROUP + i)
+ .collect(Collectors.toMap(Function.identity(),
this::buildConsumerGroupExecutor));
- String[] cgcArgs = new String[]{"--bootstrap-server",
bootstrapServers(listenerName()), "--delete", "--all-groups"};
+ String[] cgcArgs = new String[]{"--bootstrap-server",
cluster.bootstrapServers(), "--delete", "--all-groups"};
ConsumerGroupCommand.ConsumerGroupService service =
getConsumerGroupService(cgcArgs);
TestUtils.waitForCondition(() ->
- new
HashSet<>(service.listConsumerGroups()).equals(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.");
+ new
HashSet<>(service.listConsumerGroups()).equals(groupNameToExecutor.keySet()) &&
+
groupNameToExecutor.keySet().stream().allMatch(checkGroupState(service,
"Stable")),
+ "The group did not initialize as expected.");
// Shutdown consumers to empty out groups
- groups.values().forEach(AbstractConsumerGroupExecutor::shutdown);
+ for (ConsumerGroupExecutor consumerGroupExecutor :
groupNameToExecutor.values()) {
+ consumerGroupExecutor.close();
+ }
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.");
+
groupNameToExecutor.keySet().stream().allMatch(checkGroupState(service,
"Empty")),
+ "The group did not become empty as expected.");
String output =
ToolsTestUtils.grabConsoleOutput(service::deleteGroups).trim();
- Set<String> expectedGroupsForDeletion = groups.keySet();
+ Set<String> expectedGroupsForDeletion = groupNameToExecutor.keySet();
Set<String> deletedGroupsGrepped =
Arrays.stream(output.substring(output.indexOf('(') + 1,
output.indexOf(')')).split(","))
- .map(str -> str.replaceAll("'",
"").trim()).collect(Collectors.toSet());
+ .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");
+ && Objects.equals(deletedGroupsGrepped,
expectedGroupsForDeletion),
+ "The consumer group(s) could not be deleted as expected");
}
- @ParameterizedTest
- @ValueSource(strings = {"zk", "kraft"})
- public void testDeleteEmptyGroup(String quorum) throws Exception {
- createOffsetsTopic(listenerName(), new Properties());
+ @ClusterTest
+ public void testDeleteEmptyGroup() throws Exception {
Review Comment:
```java
try (ConsumerGroupExecutor consumerGroupExecutor =
buildConsumerGroupExecutor(GROUP)) {
String[] cgcArgs = new String[]{"--bootstrap-server",
cluster.bootstrapServers(), "--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."
);
consumerGroupExecutor.close();
TestUtils.waitForCondition(
() ->
Objects.equals(service.collectGroupState(GROUP).state, "Empty"),
"The group did not become empty as expected."
);
Map<String, Throwable> result = new HashMap<>();
String output = ToolsTestUtils.grabConsoleOutput(() ->
result.putAll(service.deleteGroups()));
assertTrue(output.contains("Deletion of requested consumer
groups ('" + GROUP + "') was successful."),
"The consumer group could not be deleted as expected");
assertTrue(result.size() == 1 && result.containsKey(GROUP) &&
result.get(GROUP) == null,
"The consumer group could not be deleted as expected");
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
