@ClusterTemplate("generator")
    public void testResetOffsetsToEarliestOnTopics(ClusterInstance cluster) throws Exception {
        for (GroupProtocol groupProtocol : cluster.supportedGroupProtocols()) {
            String group = generateRandomGroupId();
            String topic1 = generateRandomTopic();
            String topic2 = generateRandomTopic();
            String[] args = buildArgsForGroup(cluster, group,
                "--topic", topic1,
                "--topic", topic2,
                "--to-earliest", "--execute");

            try (Admin admin = cluster.createAdminClient();
                 ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(args)) {
                admin.createTopics(asList(new NewTopic(topic1, 1, (short) 1),
                        new NewTopic(topic2, 1, (short) 1))).all().get();

                produceConsumeAndShutdown(cluster, topic1, group, 1, groupProtocol);
                produceConsumeAndShutdown(cluster, topic2, group, 1, groupProtocol);

                TopicPartition tp1 = new TopicPartition(topic1, 0);
                TopicPartition tp2 = new TopicPartition(topic2, 0);

                Map<TopicPartition, Long> allResetOffsets = toOffsetMap(resetOffsets(service).get(group));
                Map<TopicPartition, Long> expMap = new HashMap<>();
                expMap.put(tp1, 0L);
                expMap.put(tp2, 0L);
                assertEquals(expMap, allResetOffsets);
                assertEquals(singletonMap(tp1, 0L), committedOffsets(cluster, topic1, group));
                assertEquals(singletonMap(tp2, 0L), committedOffsets(cluster, topic2, group));

                admin.deleteTopics(asList(topic1, topic2)).all().get();
            }
        }
    }
Test cases by:
- Non-existing consumer group
- One for each scenario, with scope=all-topics
- scope=one topic, scenario=to-earliest
- scope=one topic+partitions, scenario=to-earliest
- scope=topics, scenario=to-earliest
- scope=topics+partitions, scenario=to-earliest
- export/import
tools/src/test/java/org/apache/kafka/tools/consumer/group/ResetConsumerGroupOffsetTest.java