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

Code before:
    private String[] buildArgsForGroup(ClusterInstance cluster, String group, String... args) {
        return buildArgsForGroups(cluster, singletonList(group), args);
    }

    private String[] buildArgsForAllGroups(ClusterInstance cluster, String... args) {
        List<String> res = new ArrayList<>(asList(basicArgs(cluster)));
        res.add("--all-groups");
        res.addAll(asList(args));
        return res.toArray(new String[0]);
    }

    @ClusterTest
    public void testResetOffsetsNotExistingGroup(ClusterInstance cluster) throws Exception {
        String topic = generateRandomTopic();
        String group = "missing.group";
        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 = 5)
    public void testResetOffsetsWithOfflinePartition(ClusterInstance cluster) throws Exception {
        String topic = generateRandomTopic();
        String group = "new.group";
        String[] args = buildArgsForGroup(cluster, group, "--to-earliest", "--execute", "--topic", topic + ":1");

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

            cluster.shutdownBroker(3);
            cluster.shutdownBroker(4);
            TestUtils.waitForCondition(() -> !cluster.aliveBrokers().keySet().containsAll(Set.of(3, 4)),
                    "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, 1)), 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);

Review comment (suggestion):
To better distinguish this test from `testResetOffsetsWithPartitionNoneLeader`, maybe rename it to `testResetOffsetsWithOfflinePartitionNotInResetTarget` or something similar.

Code after:
    private String[] buildArgsForGroup(ClusterInstance cluster, String group, String... args) {
        return buildArgsForGroups(cluster, singletonList(group), args);
    }

    private String[] buildArgsForAllGroups(ClusterInstance cluster, String... args) {
        List<String> res = new ArrayList<>(asList(basicArgs(cluster)));
        res.add("--all-groups");
        res.addAll(asList(args));
        return res.toArray(new String[0]);
    }

    @ClusterTest
    public void testResetOffsetsNotExistingGroup(ClusterInstance cluster) throws Exception {
        String topic = generateRandomTopic();
        String group = "missing.group";
        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");