@ClusterTemplate("generator")
    public void testResetOffsetsAllTopicsAllGroups(ClusterInstance cluster) throws Exception {
        for (GroupProtocol groupProtocol : cluster.supportedGroupProtocols()) {
            String groupId = generateRandomGroupId();
            String topicId = generateRandomTopic();

            String[] args = buildArgsForAllGroups(cluster, "--all-topics", "--to-offset", "50");
            List<String> topics = generateIds(groupId);
            List<String> groups = generateIds(topicId);
            topics.forEach(topic -> produceMessages(cluster, topic, 100));

            for (String topic : topics) {
                for (String group : groups) {
                    try (AutoCloseable consumerGroupCloseable =
                                 consumerGroupClosable(cluster, 3, topic, group, groupProtocol)) {
                        awaitConsumerProgress(cluster, topic, group, 100);
                    }
                }
            }

            resetAndAssertOffsets(cluster, args, 50, true, topics);
            resetAndAssertOffsets(cluster, addTo(args, "--dry-run"),
                    50, true, topics);
            resetAndAssertOffsets(cluster, addTo(args, "--execute"),
                    50, false, topics);

            try (Admin admin = cluster.createAdminClient()) {
                admin.deleteConsumerGroups(groups).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