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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
