@ClusterTemplate("generator")
    public void testResetOffsetsExistingTopicAllGroups(ClusterInstance cluster) throws Exception {
        for (GroupProtocol groupProtocol : cluster.supportedGroupProtocols()) {
            String topic = generateRandomTopic();
            String[] args = buildArgsForAllGroups(cluster, "--topic", topic, "--to-offset", "50");

            produceMessages(cluster, topic, 100);
            for (int i = 1; i <= 3; i++) {
                String group = generateRandomGroupId();
                try (AutoCloseable consumerGroupCloseable =
                             consumerGroupClosable(cluster, 1, topic, group, groupProtocol)) {
                    awaitConsumerProgress(cluster, topic, group, 100L);
                }
            }
            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));
        }
    }
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