Repository: apache/kafka
File: tools/src/test/java/org/apache/kafka/tools/consumer/group/ResetConsumerGroupOffsetTest.java (Java)

Code before:
        String[] args = buildArgsForGroup(cluster, group, "--all-topics", "--to-current", "--execute");

        try (ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(args)) {
            Map<TopicPartition, OffsetAndMetadata> resetOffsets = service.resetOffsets().get(group);
            assertTrue(resetOffsets.isEmpty());
            assertTrue(committedOffsets(cluster, topic, group).isEmpty());
        }
    }

    @ClusterTest(
        brokers = 3,
        serverProperties = {
            @ClusterConfigProperty(key = OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "2")
        }
    )
    public void testResetOffsetsWithOfflinePartitionNotInResetTarget(ClusterInstance cluster) throws Exception {
        String topic = generateRandomTopic();
        String group = "new.group";
        String[] args = buildArgsForGroup(cluster, group, "--to-earliest", "--execute", "--topic", topic + ":0");

        try (Admin admin = cluster.admin(); ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(args)) {
            admin.createTopics(List.of(new NewTopic(topic, Map.of(0, List.of(0), 1, List.of(1)))));
            cluster.waitTopicCreation(topic, 2);

            cluster.shutdownBroker(1);
            TestUtils.waitForCondition(() -> !cluster.aliveBrokers().containsKey(1),
                    "The cluster did not shut down the broker within the expected time.");

            Map<TopicPartition, OffsetAndMetadata> resetOffsets = service.resetOffsets().get(group);
            assertEquals(Set.of(new TopicPartition(topic, 0)), resetOffsets.keySet());
        }
    }

    @ClusterTest
    public void testResetOffsetsExistingTopic(ClusterInstance cluster) {
        String topic = generateRandomTopic();
        String group = "new.group";
        String[] args = buildArgsForGroup(cluster, group, "--topic", topic, "--to-offset", "50");

        produceMessages(cluster, topic, 100);
        resetAndAssertOffsets(cluster, args, 50, true, singletonList(topic));
        resetAndAssertOffsets(cluster, addTo(args, "--dry-run"),
                50, true, singletonList(topic));
        resetAndAssertOffsets(cluster, addTo(args, "--execute"),
                50, false, singletonList(topic));
    }

    @ClusterTest
    public void testResetOffsetsExistingTopicSelectedGroups(ClusterInstance cluster) throws Exception {
        for (GroupProtocol groupProtocol : cluster.supportedGroupProtocols()) {
            String topic = generateRandomTopic();

Review comment (suggestion):
According to https://github.com/apache/kafka/pull/20168/files#r2228677312, I understand that shutdown() already waits until the shutdown is complete. So this can be removed as well.

Code after:
        String[] args = buildArgsForGroup(cluster, group, "--all-topics", "--to-current", "--execute");

        try (ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(args)) {
            Map<TopicPartition, OffsetAndMetadata> resetOffsets = service.resetOffsets().get(group);
            assertTrue(resetOffsets.isEmpty());
            assertTrue(committedOffsets(cluster, topic, group).isEmpty());
        }
    }

    @ClusterTest(
        brokers = 2,
        serverProperties = {
            @ClusterConfigProperty(key = OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "2"),
        }
    )
    public void testResetOffsetsWithOfflinePartitionNotInResetTarget(ClusterInstance cluster) throws Exception {
        String topic = generateRandomTopic();
        String group = "new.group";
        String[] args = buildArgsForGroup(cluster, group, "--to-earliest", "--execute", "--topic", topic + ":0");

        try (Admin admin = cluster.admin(); ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(args)) {
            admin.createTopics(List.of(new NewTopic(topic, Map.of(0, List.of(0), 1, List.of(1)))));
            cluster.waitTopicCreation(topic, 2);

            cluster.shutdownBroker(1);

            Map<TopicPartition, OffsetAndMetadata> resetOffsets = service.resetOffsets().get(group);
            assertEquals(Set.of(new TopicPartition(topic, 0)), resetOffsets.keySet());
        }
    }

    @ClusterTest
    public void testResetOffsetsExistingTopic(ClusterInstance cluster) {
        String topic = generateRandomTopic();
        String group = "new.group";
        String[] args = buildArgsForGroup(cluster, group, "--topic", topic, "--to-offset", "50");

        produceMessages(cluster, topic, 100);
        resetAndAssertOffsets(cluster, args, 50, true, singletonList(topic));
        resetAndAssertOffsets(cluster, addTo(args, "--dry-run"),
                50, true, singletonList(topic));
        resetAndAssertOffsets(cluster, addTo(args, "--execute"),
                50, false, singletonList(topic));
    }

    @ClusterTest
    public void testResetOffsetsExistingTopicSelectedGroups(ClusterInstance cluster) throws Exception {
        for (GroupProtocol groupProtocol : cluster.supportedGroupProtocols()) {
            String topic = generateRandomTopic();

            produceMessages(cluster, topic, 100);