chia7712 commented on code in PR #15766: URL: https://github.com/apache/kafka/pull/15766#discussion_r1585460093
########## tools/src/test/java/org/apache/kafka/tools/consumer/group/DeleteConsumerGroupsTest.java: ########## @@ -17,279 +17,332 @@ 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.GroupProtocol; import org.apache.kafka.clients.consumer.RangeAssignor; +import org.apache.kafka.common.ConsumerGroupState; 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.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.util.Arrays; +import java.util.HashMap; import java.util.HashSet; 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.function.Predicate; import java.util.stream.Collectors; import java.util.stream.IntStream; +import static java.util.Collections.emptyMap; +import static java.util.Collections.singletonList; +import static java.util.Collections.singletonMap; +import static org.apache.kafka.clients.consumer.GroupProtocol.CLASSIC; +import static org.apache.kafka.clients.consumer.GroupProtocol.CONSUMER; +import static org.apache.kafka.common.ConsumerGroupState.EMPTY; +import static org.apache.kafka.common.ConsumerGroupState.STABLE; +import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.NEW_GROUP_COORDINATOR_ENABLE_CONFIG; +import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG; +import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG; 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"}; - assertThrows(OptionException.class, () -> getConsumerGroupService(cgcArgs)); - } - - @ParameterizedTest - @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 = 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"); +@ExtendWith(value = ClusterTestExtensions.class) +@ClusterTestDefaults(clusterType = Type.ALL, serverProperties = { + @ClusterConfigProperty(key = OFFSETS_TOPIC_PARTITIONS_CONFIG, value = "1"), + @ClusterConfigProperty(key = OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "1"), + @ClusterConfigProperty(key = NEW_GROUP_COORDINATOR_ENABLE_CONFIG, value = "true") Review Comment: @frankvicky @FrankYang0529 Could you please address @lianetm comments? This class and `DeleteOffsetsConsumerGroupCommandIntegrationTest` need to test "LegacyConsumer + NEW_GROUP_COORDINATOR_ENABLE_CONFIG=false (`GroupCoordinatorAdapter`)" We can address that by `ClusterTemplate`. For example: ```java private static void generator(ClusterGenerator clusterGenerator) { Map<String, String> serverProperties = new HashMap<>(); serverProperties.put(GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG, "1"); serverProperties.put(GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, "1"); serverProperties.put(GroupCoordinatorConfig.NEW_GROUP_COORDINATOR_ENABLE_CONFIG, "true"); // case 0: zk clusterGenerator.accept(ClusterConfig.defaultBuilder().setType(Type.ZK).setServerProperties(serverProperties).build()); // case 1: kraft + legacy coordinator clusterGenerator.accept(ClusterConfig.defaultBuilder().setType(Type.KRAFT).setServerProperties(serverProperties).build()); // case 2: combined kraft + legacy coordinator clusterGenerator.accept(ClusterConfig.defaultBuilder().setType(Type.CO_KRAFT).setServerProperties(serverProperties).build()); serverProperties.put(GroupCoordinatorConfig.NEW_GROUP_COORDINATOR_ENABLE_CONFIG, "true"); // case 3: kraft + new group coordinator clusterGenerator.accept(ClusterConfig.defaultBuilder().setType(Type.KRAFT).setName("newGroupCoordinator").setServerProperties(serverProperties).build()); // case 4: combined kraft + new group coordinator clusterGenerator.accept(ClusterConfig.defaultBuilder().setType(Type.CO_KRAFT).setName("newGroupCoordinator").setServerProperties(serverProperties).build()); } ``` -- 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