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

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

                produceConsumeAndShutdown(cluster, topic, group, 2, groupProtocol);
                Map<TopicPartition, Long> priorCommittedOffsets = committedOffsets(cluster, topic, group);
                TopicPartition tp0 = new TopicPartition(topic, 0);
                TopicPartition tp1 = new TopicPartition(topic, 1);
                Map<TopicPartition, Long> expectedOffsets = new HashMap<>();
                expectedOffsets.put(tp0, priorCommittedOffsets.get(tp0));
                expectedOffsets.put(tp1, 0L);
                resetAndAssertOffsetsCommitted(cluster, service, expectedOffsets, topic);

                admin.deleteTopics(singleton(topic)).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