KAFKA-16593: Rewrite DeleteConsumerGroupsTest by ClusterTestExtensions#15766

frankvicky · 2024-04-21T06:29:00Z

Rewrite DeleteConsumerGroupsTest by ClusterTestExtensions.

chia7712

@frankvicky thanks for this contribution. some comments left. PTAL

chia7712 · 2024-04-21T16:58:02Z

-                result.containsKey(missingGroup) &&
-                result.get(missingGroup).getMessage().contains(Errors.GROUP_ID_NOT_FOUND.message()),
-            "The consumer group deletion did not work as expected");
+    class AbstractConsumerGroupExecutor {


those classes can be static class.

chia7712 · 2024-04-21T16:58:22Z

+            assert configured : "Must call configure before use";
+            try {
+                subscribe();
+                while (true) {


Could we add a flag to break the loop when calling shutdown?

chia7712 · 2024-04-21T16:59:27Z

+    @ClusterTest
+    public void testDeleteWithTopicOption() {
+        try (Admin admin = cluster.createAdminClient()) {
+            admin.createTopics(buildSingletonTestTopic());


please call get to make sure the request is completed.

chia7712 · 2024-04-21T17:02:40Z

-    public void testDeleteWithUnrecognizedNewConsumerOption(String quorum) {
-        String[] cgcArgs = new String[]{"--new-consumer", "--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", GROUP};
-        assertThrows(OptionException.class, () -> getConsumerGroupService(cgcArgs));
+    class ConsumerGroupExecutor extends AbstractConsumerGroupExecutor {


AbstractConsumerGroupExecutor is extended by ConsumerGroupExecutor only, so we can merge them together.

chia7712 · 2024-04-21T17:04:21Z

+            executor.submit(consumerThread);
+        }
+
+        void shutdown() {


We can change this to close and make AbstractConsumerGroupExecutor extend Closeable. That allow us to use try-with-resources

In this usage scenario, many test cases involve manually invoking the shutdown method and subsequently checking the status of resource closure. In this case, do I still need to implement the AutoCloseable interface?

The first option is to keep it as it is.
The second option is to implement AutoCloseable, but still manually call the close method.
The third option is to implement AutoCloseable, but remove the tests for resource release.

chia7712 · 2024-04-21T17:04:52Z

-        createOffsetsTopic(listenerName(), new Properties());
-        String missingGroup = "missing.group";
+@ExtendWith(value = ClusterTestExtensions.class)
+@ClusterTestDefaults(clusterType = Type.ALL, brokers = 3, serverProperties = {


We reduce the number of partitions/replicas so do we need 3 brokers?

chia7712 · 2024-04-22T13:57:46Z

+                Admin admin = cluster.createAdminClient();
+                ConsumerGroupExecutor consumerGroupExecutor = buildConsumerGroupExecutor(GROUP)
+        ) {
+            createAndAwaitTestTopic(admin);


Kafka will auto-create the topic when the topic is nonexistent, so you don't need to call createAndAwaitTestTopic

chia7712

@frankvicky thanks for updated PR. I notice that there are many similar test cases. Please try to merge them to simplify this test

chia7712 · 2024-04-22T15:55:17Z

-        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 {


Could we merge testDeleteCmdNonEmptyGroup into testDeleteNonEmptyGroup? For example:

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 = new HashMap<>(); String output = ToolsTestUtils.grabConsoleOutput(() -> result.putAll(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 + ")"); 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 + ")"); }

chia7712 · 2024-04-22T15:56:12Z

-    public void testDeleteEmptyGroup(String quorum) throws Exception {
-        createOffsetsTopic(listenerName(), new Properties());
+    @ClusterTest
+    public void testDeleteEmptyGroup() throws Exception {


ditto that testDeleteEmptyGroup and testDeleteCmdEmptyGroup can get merged

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"); }

lianetm · 2024-04-22T19:59:59Z

-            "The group did not initialize as expected."
-        );
+            TestUtils.waitForCondition(
+                    () -> service.listConsumerGroups().contains(GROUP) && Objects.equals(service.collectGroupState(GROUP).state, "Stable"),


There is a ConsumerGroupState enum with all the group states. Would be sensible to use it instead of the literal "Stable" just to avoid breaks/duplicated work in the future if states get updated (ditto for all other references to literal state names)

lianetm · 2024-04-22T20:13:50Z

+        void configure(Properties props) {
+            super.configure(props);
+            props.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, groupProtocol);
+            if (groupProtocol.toUpperCase(Locale.ROOT).equals(GroupProtocol.CONSUMER.toString())) {


This could be simplified to if (groupProtocol == GroupProtocol.CONSUMER) if we work with the GroupProtocol object instead of a string throughout the test. We truly only need the string when passing it into the consumer prop GROUP_PROTOCOL_CONFIG above.

lianetm · 2024-04-22T21:01:38Z

+        return new ConsumerGroupExecutor(cluster.bootstrapServers(),
+                1,
+                null != group ? group : GROUP,
+                GroupProtocol.CLASSIC.name,


The consumer group command supports deleting CONSUMER groups, not only CLASSIC ones, so shouldn't we parametrize this test to make sure checks deleting both group types?

Not sure if I'm missing something or that's being tracked in a different jira maybe @dajac? (I can't find it)

agree to add test for CONSUMER. The origin test case does not include that, and we can complete that in this PR.

@frankvicky It seems to me the following changes can be added.

add @ClusterConfigProperty(key = "group.coordinator.new.enable", value = "true")

use cluster.isKRaftTest() to avoid using GroupProtocol.CONSUMER in zk mode

we will also need @ClusterConfigProperty(key = "group.coordinator.rebalance.protocols", value = "consumer,classic") (along with group.coordinator.new.enable=true) to run with the new protocol

we will also need @ClusterConfigProperty(key = "group.coordinator.rebalance.protocols", value = "consumer,classic") (along with group.coordinator.new.enable=true )

It seems all we need is to set one of them. see

kafka/core/src/main/scala/kafka/server/KafkaConfig.scala

Line 1170 in 1b301b3

val isNewGroupCoordinatorEnabled = getBoolean(GroupCoordinatorConfig.NEW_GROUP_COORDINATOR_ENABLE_CONFIG) ||

You're totally right, I was wrongfully thinking it was also needed (and always using it like that), mainly because the list of supported protocols defaults to classic only, but agree, not needed.

frankvicky · 2024-04-23T15:35:15Z

Hi @lianetm, @chia7712
Thanks for the suggestions, I have addressed the comments. 😀

chia7712

@frankvicky thanks for updated PR. a couple of comments left.

chia7712 · 2024-04-23T16:07:43Z

+    }

-        cgcArgs = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--delete", "--group", GROUP, "--group", missingGroup};
+    static abstract class AbstractConsumerRunnable implements Runnable {


As AbstractConsumerRunnable has only one sub-class, we can merge them into to one class

chia7712 · 2024-04-23T16:07:58Z

+    static abstract class AbstractConsumerRunnable implements Runnable {
+        final String broker;
+        final String groupId;
+        final Optional<Properties> customPropsOpt;


if customPropsOpt is always, please remove it.

chia7712 · 2024-04-23T16:08:18Z

+        final String topic;
+        final String groupProtocol;
+        final String strategy;
+        final Optional<String> remoteAssignor;


chia7712 · 2024-04-23T16:12:30Z

@frankvicky @m1a2st It seems your PR (#15766 and #15779) need a consumer running in background. Hence, we can consider moving AbstractConsumerGroupExecutor/ConsumerGroupExecutor to a individual java file. WDYT?

lianetm · 2024-04-23T16:23:41Z

@chia7712 's comment makes sense to me, and heads-up, similar classes are already defined in ConsumerGroupCommandTest, so probably sensible to take a look at all and consider consolidating

frankvicky · 2024-04-23T16:28:15Z

@frankvicky @m1a2st It seems your PR (#15766 and #15779) need a consumer running in background. Hence, we can consider moving AbstractConsumerGroupExecutor/ConsumerGroupExecutor to a individual java file. WDYT?

It sounds good. 😀

lianetm · 2024-04-23T18:41:59Z

+    public ConsumerGroupExecutor(
+            String broker,
+            int numConsumers,
+            String groupId,
+            String groupProtocol,
+            String topic,
+            String strategy,
+            Optional<String> remoteAssignor,
+            Optional<Properties> customPropsOpt,
+            boolean syncCommit


We have 2 well defined group types, classic and consumer, with different creation params, and having them all together seems a bit confusing (ex. having assignment strategy along with remote assignor)...not very intuitive when creating a new instance (that I expect several other test files will do as they get migrated). So wonder if it would maybe be clearer to split them into static builders that would take only the params they need? like:

public static ConsumerGroupExecutor forConsumerGroup( String broker, int numConsumers, String groupId, String topic, Optional<String> remoteAssignor, Optional<Properties> customPropsOpt, boolean syncCommit) { } public static ConsumerGroupExecutor forClassicGroup( String broker, int numConsumers, String groupId, String topic, String strategy, Optional<Properties> customPropsOpt, boolean syncCommit) { }

Indeed, too many parameters can be very confusing for users. I will make the changes according to your suggestion. Thank you! 😀

lianetm · 2024-04-23T18:43:43Z

+            String groupId,
+            String groupProtocol,
+            String topic,
+            String strategy,


maybe rename this to assignmentStrategy for clarity on what it is

Thank you, this make sense. 😄

lianetm · 2024-04-23T18:52:23Z

+            String strategy,
+            Optional<String> remoteAssignor,
+            Optional<Properties> customPropsOpt,
+            boolean syncCommit


The syncCommit param seems to always be false. If that's the case what's the purpose of it? (is it with the intention of using this same common ConsumerGroupExecutor class from the ConsumerGroupCommandTest too?)

Hi @lianetm
Yes, your assumption is correct. I'm pretty new to being a Committer myself, so I aimed to complete this migration with minimal alterations.

Makes sense, let's just make sure we have a follow-up Jira to update the ConsumerGroupCommandTest then, so we don't leave the duplicated ConsumerGroupExecutor we currently have.

chia7712

@frankvicky thanks for updated PR. a couple of style-related comments. PTAL

chia7712 · 2024-04-24T13:19:40Z

+import static org.apache.kafka.clients.consumer.ConsumerConfig.GROUP_REMOTE_ASSIGNOR_CONFIG;
+import static org.apache.kafka.common.GroupType.CONSUMER;
+
+public class ConsumerRunnable implements Runnable {


This class is used by ConsumerGroupExecutor so we can move it to be a static class within ConsumerGroupExecutor

chia7712 · 2024-04-24T13:20:31Z

+        }
+    }
+
+    void subscribe() {


We don't need this method as it is used by run only

chia7712 · 2024-04-24T13:22:08Z

+
+    @Override
+    public void run() {
+        assert configured : "Must call configure before use";


We expect configure must be called so it would be nice to move it to the constructor. It means we should create the consumer in constructor.

chia7712

@frankvicky thanks for updated PR. a couple of comments left. PTAL

chia7712 · 2024-04-25T19:26:50Z

+import static org.apache.kafka.clients.consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG;
+import static org.apache.kafka.common.GroupType.CONSUMER;
+
+public class ConsumerGroupExecutor implements AutoCloseable {


Remove the public to make it be package-private

chia7712 · 2024-04-25T19:27:46Z

+        });
+    }
+
+    public static ConsumerGroupExecutor buildConsumerGroup(String brokerAddress,


ditto: remove public

chia7712 · 2024-04-25T19:27:49Z

+        );
+    }
+
+    public static ConsumerGroupExecutor buildClassicGroup(String brokerAddress,


ditto: remove public

chia7712 · 2024-04-25T19:28:33Z

+        }
+    }
+
+    static class ConsumerRunnable implements Runnable {


add private since it is not exposed to callers

chia7712 · 2024-04-25T19:28:38Z

+        private boolean configured = false;
+        private volatile boolean isShutdown = false;
+
+        public ConsumerRunnable(String brokerAddress,


chia7712 · 2024-04-25T19:30:52Z

+    }
+
+    static class ConsumerRunnable implements Runnable {
+        private final String brokerAddress;


those inner variables are not used after construction, so we can remove them

chia7712 · 2024-04-25T19:32:02Z

+                                String topic,
+                                String assignmentStrategy,
+                                Optional<String> remoteAssignor,
+                                Optional<Properties> customConfigs,


Map<String, Object> is more suitable in this case. The empty map can be equal to Optional.empty but it gets more graceful.

chia7712 · 2024-04-25T19:33:14Z

+            this.assignmentStrategy = assignmentStrategy;
+            this.remoteAssignor = remoteAssignor;
+
+            this.configure();


We don't need this method, since it is used only once. For example:

Map<String, Object> configs = new HashMap<>(); configs.put(BOOTSTRAP_SERVERS_CONFIG, brokerAddress); configs.put(GROUP_ID_CONFIG, groupId); configs.put(KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); configs.put(VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); configs.put(GROUP_PROTOCOL_CONFIG, groupProtocol); if (Objects.equals(groupProtocol, CONSUMER.toString())) { remoteAssignor.ifPresent(assignor -> configs.put(GROUP_REMOTE_ASSIGNOR_CONFIG, assignor)); } else { configs.put(PARTITION_ASSIGNMENT_STRATEGY_CONFIG, assignmentStrategy); } configs.putAll(customConfigs); consumer = new KafkaConsumer<>(configs);

chia7712

@frankvicky thanks for updated PR. two major comments left. PTAL

chia7712 · 2024-04-26T01:36:02Z

-                                                           Optional<String> remoteAssignor,
-                                                           Optional<Properties> customConfigs,
-                                                           boolean syncCommit) {
+    static ConsumerGroupExecutor buildConsumerGroup(String brokerAddress,


Could you move those static methods up?

chia7712 · 2024-04-26T02:11:06Z

+                configs.put(PARTITION_ASSIGNMENT_STRATEGY_CONFIG, assignmentStrategy);
            }
+            configs.putAll(customConfigs);
+            consumer = new KafkaConsumer<>(configs);


Could you make sure all consumers get closed even though one of consumer gets error? We encountered resource leaks before and it makes our CI unstable. For example:

private static AutoCloseable run( String brokerAddress, int numberOfConsumers, String groupId, String groupProtocol, String topic, String assignmentStrategy, Optional<String> remoteAssignor, Map<String, Object> customConfigs, boolean syncCommit ) { Queue<Consumer<byte[], byte[]>> consumers = consumers(IntStream.range(0, numberOfConsumers).mapToObj(ignored -> { Map<String, Object> configs = new HashMap<>(); configs.put(BOOTSTRAP_SERVERS_CONFIG, brokerAddress); configs.put(GROUP_ID_CONFIG, groupId); configs.put(GROUP_PROTOCOL_CONFIG, groupProtocol); if (Objects.equals(groupProtocol, CONSUMER.toString())) { remoteAssignor.ifPresent(assignor -> configs.put(GROUP_REMOTE_ASSIGNOR_CONFIG, assignor)); } else { configs.put(PARTITION_ASSIGNMENT_STRATEGY_CONFIG, assignmentStrategy); } configs.putAll(customConfigs); return configs; }).collect(Collectors.toList())); AtomicBoolean closed = new AtomicBoolean(false); ExecutorService service = Executors.newFixedThreadPool(consumers.size()); final AutoCloseable closeable = () -> { closed.set(true); consumers.forEach(c -> Utils.closeQuietly(c, "close consumer")); service.shutdownNow(); service.awaitTermination(1, TimeUnit.MINUTES); }; try { while (!consumers.isEmpty()) { Consumer<byte[], byte[]> consumer = consumers.poll(); service.submit(() -> { try { consumer.subscribe(singleton(topic)); while (closed.get()) { consumer.poll(Duration.ofMillis(Long.MAX_VALUE)); if (syncCommit) consumer.commitSync(); } } catch (WakeupException | InterruptException e) { // OK } finally { consumer.close(); } }); } return closeable; } catch (Throwable e) { Utils.closeQuietly(closeable, "release consumer"); throw e; } } private static Queue<Consumer<byte[], byte[]>> consumers(List<Map<String, Object>> allConfigs) { Queue<Consumer<byte[], byte[]>> consumers = new LinkedList<>(); try { allConfigs.forEach(configs -> consumers.add(new KafkaConsumer<>(configs, new ByteArrayDeserializer(), new ByteArrayDeserializer()))); return consumers; } catch (Throwable e) { consumers.forEach(c -> Utils.closeQuietly(c, "close consumer")); throw e; } }

chia7712

@frankvicky thanks for updated PR. there are two major comments. PTAL

chia7712 · 2024-04-26T23:30:59Z

+                executor.execute(() -> {
+                    try {
+                        consumer.subscribe(singleton(topic));
+                        while (closed.get()) {


while (!closed.get())

chia7712 · 2024-04-26T23:31:47Z

+        @ClusterConfigProperty(key = NEW_GROUP_COORDINATOR_ENABLE_CONFIG, value = "true")
+})
+public class DeleteConsumerGroupsTest {
+    private static final String TOPIC = "foo";


As we run a loop in each test case, it would be nice to create different topic/group for them. please take a look at #15802

chia7712 · 2024-04-28T17:19:02Z

I file https://issues.apache.org/jira/browse/KAFKA-16639 to trace the potential bug about AsyncConsumer. Please wait for that issue. If the issue is too hard to resolve, we can remove the test case about new consumer from this PR

chia7712 · 2024-04-29T06:16:40Z

+                            if (syncCommit)
+                                consumer.commitSync();
+                        }
+                    } catch (WakeupException | InterruptException e) {


Could you clear the interrupt status to avoid impacting the consumer#close? for example:

} catch (WakeupException e) { // OK } catch (InterruptException e) { // OK with reset interrupt Thread.interrupted(); } finally { consumer.close(); }

chia7712 · 2024-04-29T17:37:47Z

+                )
+                .collect(Collectors.toList());
+
+        Queue<KafkaConsumer<String, String>> kafkaConsumers = buildConsumersSafely(allConfigs);


Maybe we can get it more simple.

private static AutoCloseable buildConsumers( String brokerAddress, int numberOfConsumers, String groupId, String groupProtocol, String topic, String assignmentStrategy, Optional<String> remoteAssignor, Map<String, Object> customConfigs, boolean syncCommit ) { List<Map<String, Object>> allConfigs = IntStream.range(0, numberOfConsumers) .mapToObj(ignored -> composeConfigs( brokerAddress, groupId, groupProtocol, assignmentStrategy, remoteAssignor, customConfigs ) ) .collect(Collectors.toList()); List<KafkaConsumer<byte[], byte[]>> kafkaConsumers = Collections.synchronizedList(new ArrayList<>()); ExecutorService executor = Executors.newFixedThreadPool(allConfigs.size()); AtomicBoolean closed = new AtomicBoolean(false); final AutoCloseable closeable = () -> releaseConsumers(closed, kafkaConsumers, executor); try { for (Map<String, Object> configs : allConfigs) { executor.execute(() -> { try(KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<>(configs)) { kafkaConsumers.add(consumer); consumer.subscribe(singleton(topic)); while (!closed.get()) { consumer.poll(Duration.ofMillis(Long.MAX_VALUE)); if (syncCommit) consumer.commitSync(); } } catch (WakeupException e) { // OK } }); } return closeable; } catch (Throwable e) { Utils.closeQuietly(closeable, "Release Consumer"); throw e; } } private static void releaseConsumers(AtomicBoolean closed, List<KafkaConsumer<byte[], byte[]>> kafkaConsumers, ExecutorService executor) throws InterruptedException { closed.set(true); kafkaConsumers.forEach(KafkaConsumer::wakeup); executor.shutdown(); executor.awaitTermination(1, TimeUnit.MINUTES); }

…sumerGroupService to try-with resource block to leverage the autocloseable.

… all scenario.

…Instance, and refactor DeleteConsumerGroupsTest.

…GroupState and refactor related test.

chia7712 · 2024-05-05T05:47:12Z

kafka/tools/src/main/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommand.java

Line 433 in 25118ce

    
           System.out.printf(format, state.group, coordinator, state.assignmentStrategy, state.state, state.numMembers);

@frankvicky Could you change state.state to state.state.name() for same output. Also, please fix the build error

chia7712

@frankvicky thanks for updated PR. two comments left.

chia7712 · 2024-05-05T06:47:22Z


                    System.out.printf(format, "GROUP", "COORDINATOR (ID)", "ASSIGNMENT-STRATEGY", "STATE", "#MEMBERS");
-                    System.out.printf(format, state.group, coordinator, state.assignmentStrategy, state.state, state.numMembers);
+                    System.out.printf(format, state.group, coordinator, state.assignmentStrategy, state.state.name(), state.numMembers);


Sorry that we should use toString() instead of name()

chia7712

@frankvicky thanks for updated PR. Some comments are left. PTAL

chia7712 · 2024-05-05T17:31:47Z

+
+                cgcArgs = new String[]{"--bootstrap-server", cluster.bootstrapServers(), "--delete", "--group", groupId, "--group", missingGroupId};
+
+                ConsumerGroupCommand.ConsumerGroupService service2 = getConsumerGroupService(cgcArgs);


Please close this service2

chia7712 · 2024-05-05T17:32:04Z

+
+                cgcArgs = new String[]{"--bootstrap-server", cluster.bootstrapServers(), "--delete", "--group", groupId, "--group", missingGroupId};
+
+                ConsumerGroupCommand.ConsumerGroupService service2 = getConsumerGroupService(cgcArgs);


chia7712 · 2024-05-05T17:33:09Z

+    @ClusterTemplate("generator")
+    public void testDeleteWithUnrecognizedNewConsumerOption() {
+        String[] cgcArgs = new String[]{"--new-consumer", "--bootstrap-server", cluster.bootstrapServers(), "--delete", "--group", getDummyGroupId()};
+        assertThrows(OptionException.class, () -> getConsumerGroupService(cgcArgs));


We expect the option parse get failed, so it would be more suitable to test ConsumerGroupCommandOptions.fromArgs directly

chia7712 · 2024-05-05T17:33:17Z

+    @ClusterTemplate("generator")
+    public void testDeleteWithTopicOption() {
+        String[] cgcArgs = new String[]{"--bootstrap-server", cluster.bootstrapServers(), "--delete", "--group", getDummyGroupId(), "--topic"};
+        assertThrows(OptionException.class, () -> getConsumerGroupService(cgcArgs));


chia7712 · 2024-05-05T17:35:12Z

+    }

-        ConsumerGroupCommand.ConsumerGroupService service2 = getConsumerGroupService(cgcArgs);
+    @ClusterTemplate("generator")


It tests only the arguments, so we don't need to create a embedded server.

chia7712 · 2024-05-05T17:38:31Z

-        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 + ")");
+    @ClusterTemplate("generator")


ditto. replace it by @Test

chia7712 · 2024-05-05T17:40:08Z

-        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");
+    public DeleteConsumerGroupsTest(ClusterInstance cluster) {


As some test cases don't require ClusterInstance, please inject ClusterInstance to method-level instead of class-level. For example:

@ClusterTemplate("generator") public void testDeleteCmdNonExistingGroup(ClusterInstance cluster)

…method level

chia7712

@frankvicky Sorry that I have left more comments. PTAL

chia7712 · 2024-05-06T11:51:19Z

    public void testClusterTemplate() {
        Assertions.assertEquals(ClusterInstance.ClusterType.ZK, clusterInstance.clusterType(),
-            "generate1 provided a Zk cluster, so we should see that here");
+                "generate1 provided a Zk cluster, so we should see that here");


chia7712 · 2024-05-06T11:51:22Z

+                "generate1 provided a Zk cluster, so we should see that here");
        Assertions.assertEquals("Generated Test", clusterInstance.config().name().orElse(""),
-            "generate1 named this cluster config, so we should see that here");
+                "generate1 named this cluster config, so we should see that here");


chia7712 · 2024-05-06T11:51:26Z

-        })
+            @ClusterTest(name = "cluster-tests-1", clusterType = Type.ZK, serverProperties = {
+                    @ClusterConfigProperty(key = "foo", value = "bar"),
+                    @ClusterConfigProperty(key = "spam", value = "eggs")


chia7712 · 2024-05-06T11:51:31Z

-        @ClusterTest(clusterType = Type.KRAFT, disksPerBroker = 2),
-        @ClusterTest(clusterType = Type.CO_KRAFT),
-        @ClusterTest(clusterType = Type.CO_KRAFT, disksPerBroker = 2)
+            @ClusterTest(clusterType = Type.ZK),


chia7712 · 2024-05-06T11:52:20Z

+    })
+    public void testNotSupportedNewGroupProtocols(ClusterInstance clusterInstance) {
+        Assertions.assertTrue(clusterInstance.supportedGroupProtocols().contains(CLASSIC));
+        Assertions.assertEquals(clusterInstance.supportedGroupProtocols().size(), 1);


the first argument is "expected" value, so it should be Assertions.assertEquals(1, clusterInstance.supportedGroupProtocols().size());

chia7712 · 2024-05-06T11:57:44Z

+
+import static java.util.Collections.singleton;
+
+class ConsumerGroupExecutor {


Other tests (for example: #15872) need both this and generator. How about renaming this to ConsumerGroupCommandTestUtils and move the implementation of generator to it?

…o it.

chia7712 · 2024-05-07T00:51:17Z

@frankvicky Could you please rebase code to trigger QA again?

chia7712

LGTM

…apache#15766) Reviewers: Chia-Ping Tsai <chia7712@gmail.com>

chia7712 reviewed Apr 21, 2024

View reviewed changes

chia7712 reviewed Apr 22, 2024

View reviewed changes

frankvicky changed the title ~~[DRAFT] KAFKA-16593: wip~~ [DRAFT] KAFKA-16593: Rewrite DeleteConsumerGroupsTest by ClusterTestExtensions Apr 22, 2024

frankvicky marked this pull request as ready for review April 22, 2024 14:40

frankvicky changed the title ~~[DRAFT] KAFKA-16593: Rewrite DeleteConsumerGroupsTest by ClusterTestExtensions~~ KAFKA-16593: Rewrite DeleteConsumerGroupsTest by ClusterTestExtensions Apr 22, 2024

chia7712 reviewed Apr 22, 2024

View reviewed changes

lianetm reviewed Apr 22, 2024

View reviewed changes

chia7712 mentioned this pull request Apr 23, 2024

KAFKA-16483: migrate DeleteOffsetsConsumerGroupCommandIntegrationTest to use ClusterTestExtensions #15679

Merged

3 tasks

chia7712 reviewed Apr 23, 2024

View reviewed changes

lianetm reviewed Apr 23, 2024

View reviewed changes

chia7712 reviewed Apr 24, 2024

View reviewed changes

frankvicky force-pushed the KAFKA-16593 branch from 872791a to f094a55 Compare April 24, 2024 13:55

chia7712 reviewed Apr 25, 2024

View reviewed changes

chia7712 reviewed Apr 26, 2024

View reviewed changes

FrankYang0529 mentioned this pull request Apr 27, 2024

MINOR: migrate ListConsumerGroupTest to use ClusterTestExtensions #15821

Merged

3 tasks

chia7712 reviewed Apr 29, 2024

View reviewed changes

frankvicky force-pushed the KAFKA-16593 branch from 3f83a5d to 6a76d6d Compare April 29, 2024 15:47

m1a2st mentioned this pull request Apr 29, 2024

KAFKA-16598 Mirgrate ResetConsumerGroupOffsetTest to new test infra #15779

Merged

chia7712 reviewed Apr 29, 2024

View reviewed changes

frankvicky force-pushed the KAFKA-16593 branch from 8824ea5 to 4883b44 Compare April 30, 2024 12:56

frankvicky added 10 commits May 5, 2024 13:00

KAFKA-16593: rearrange class member position.

f8d77b7

KAFKA-16593: wip

3322ed7

KAFKA-16593: wip

9db7e0b

KAFKA-16593: wip

a03fca2

KAFKA-16593: Add Thread.interrupted() method to avoid flaky. Move Con…

806eef2

…sumerGroupService to try-with resource block to leverage the autocloseable.

KAFKA-16593: Refactor ConsumerGroupExecutor.

299e1ab

KAFKA-16593: Replace ClusterTestDefault with ClusterTemplate to cover…

759e74d

… all scenario.

KAFKA-16593: Refactor buildConsumers function with Supplier.

285aafe

KAFKA-16593: Fix the bug that don't utilize thread pool.

05f6ded

KAFKA-16593: Add a default method "supportedGroupProtocol" to Cluster…

ff9a39e

…Instance, and refactor DeleteConsumerGroupsTest.

frankvicky force-pushed the KAFKA-16593 branch from 1eb9811 to ff9a39e Compare May 5, 2024 05:23

KAFKA-16593: Migrate type of GroupState.state from String to Consumer…

e5c9a6d

…GroupState and refactor related test.

chia7712 reviewed May 5, 2024

View reviewed changes

Comment thread core/src/test/java/kafka/test/ClusterInstance.java

KAFKA-16593: Fix build error and ConsumerGroupCommand output

e8ce3b6

chia7712 reviewed May 5, 2024

View reviewed changes

KAFKA-16593: Add test for supportedGroupProtocols.

4f9a9b1

chia7712 reviewed May 5, 2024

View reviewed changes

FrankYang0529 mentioned this pull request May 6, 2024

MINOR: use classic consumer with ZK mode for DeleteOffsetsConsumerGroupCommandIntegrationTest #15872

Merged

3 tasks

KAFKA-16593: Change injection of ClusterInstance from class level to …

a960105

…method level

chia7712 reviewed May 6, 2024

View reviewed changes

frankvicky and others added 3 commits May 6, 2024 20:18

KAFKA-16593: Rename ConsumerGroupExecutor and move generator method t…

6d8d788

…o it.

Merge branch 'apache:trunk' into KAFKA-16593

2550ac7

KAFKA-16593: Adjust indent.

32054e6

Merge branch 'refs/heads/trunk' into KAFKA-16593

004e8cc

chia7712 approved these changes May 7, 2024

View reviewed changes

chia7712 merged commit 0de3b7c into apache:trunk May 7, 2024

chia7712 mentioned this pull request May 7, 2024

Kafka-16668: Add tags support in ClusterTestExtension #15861

Merged

3 tasks

gongxuanzhang pushed a commit to gongxuanzhang/kafka that referenced this pull request Jun 12, 2024

KAFKA-16593 Rewrite DeleteConsumerGroupsTest by ClusterTestExtensions (…

7333a24

…apache#15766) Reviewers: Chia-Ping Tsai <chia7712@gmail.com>


		cgcArgs = new String[]{"--bootstrap-server", cluster.bootstrapServers(), "--delete", "--group", groupId, "--group", missingGroupId};

		ConsumerGroupCommand.ConsumerGroupService service2 = getConsumerGroupService(cgcArgs);


		import static java.util.Collections.singleton;

		class ConsumerGroupExecutor {

Conversation

frankvicky commented Apr 21, 2024

Uh oh!

chia7712 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

chia7712 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lianetm Apr 22, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lianetm Apr 23, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

frankvicky commented Apr 23, 2024

Uh oh!

chia7712 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

chia7712 commented Apr 23, 2024

Uh oh!

lianetm commented Apr 23, 2024

Uh oh!

frankvicky commented Apr 23, 2024

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lianetm Apr 22, 2024 •

edited

Loading

lianetm Apr 23, 2024 •

edited

Loading