@ClusterTemplate("generator")
    // This one deals with universal CSV export/import file format "group,topic,partition,offset",
    // supporting multiple --group args or --all-groups arg
    public void testResetOffsetsExportImportPlan(ClusterInstance cluster) throws Exception {
        for (GroupProtocol groupProtocol : cluster.supportedGroupProtocols()) {
            String group1 = generateRandomGroupId();
            String group2 = generateRandomGroupId();
            String topic1 = generateRandomTopic();
            String topic2 = generateRandomTopic();

            TopicPartition t1p0 = new TopicPartition(topic1, 0);
            TopicPartition t1p1 = new TopicPartition(topic1, 1);
            TopicPartition t2p0 = new TopicPartition(topic2, 0);
            TopicPartition t2p1 = new TopicPartition(topic2, 1);
            String[] cgcArgs = buildArgsForGroups(cluster, asList(group1, group2),
                "--all-topics", "--to-offset", "2", "--export");
            File file = TestUtils.tempFile("reset", ".csv");

            try (Admin admin = cluster.createAdminClient();
                 ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(cgcArgs)) {

                admin.createTopics(asList(new NewTopic(topic1, 2, (short) 1),
                        new NewTopic(topic2, 2, (short) 1))).all().get();

                produceConsumeAndShutdown(cluster, topic1, group1, 1, groupProtocol);
                produceConsumeAndShutdown(cluster, topic2, group2, 1, groupProtocol);

                awaitConsumerGroupInactive(service, group1);
                awaitConsumerGroupInactive(service, group2);

                Map<String, Map<TopicPartition, OffsetAndMetadata>> exportedOffsets = service.resetOffsets();

                writeContentToFile(file, service.exportOffsetsToCsv(exportedOffsets));

                Map<TopicPartition, Long> exp1 = new HashMap<>();
                exp1.put(t1p0, 2L);
                exp1.put(t1p1, 2L);
                Map<TopicPartition, Long> exp2 = new HashMap<>();
                exp2.put(t2p0, 2L);
                exp2.put(t2p1, 2L);

                assertEquals(exp1, toOffsetMap(exportedOffsets.get(group1)));
                assertEquals(exp2, toOffsetMap(exportedOffsets.get(group2)));

                // Multiple --group's offset import
                String[] cgcArgsExec = buildArgsForGroups(cluster, asList(group1, group2),
                        "--all-topics",
                        "--from-file", file.getCanonicalPath(), "--dry-run");
                try (ConsumerGroupCommand.ConsumerGroupService serviceExec = getConsumerGroupService(cgcArgsExec)) {
                    Map<String, Map<TopicPartition, OffsetAndMetadata>> importedOffsets = serviceExec.resetOffsets();
                    assertEquals(exp1, toOffsetMap(importedOffsets.get(group1)));
                    assertEquals(exp2, toOffsetMap(importedOffsets.get(group2)));
                }

                // Single --group offset import using "group,topic,partition,offset" csv format
                String[] cgcArgsExec2 = buildArgsForGroup(cluster, group1, "--all-topics",
                        "--from-file", file.getCanonicalPath(), "--dry-run");
                try (ConsumerGroupCommand.ConsumerGroupService serviceExec2 = getConsumerGroupService(cgcArgsExec2)) {
                    Map<String, Map<TopicPartition, OffsetAndMetadata>> importedOffsets2 = serviceExec2.resetOffsets();
                    assertEquals(exp1, toOffsetMap(importedOffsets2.get(group1)));
                }

                admin.deleteTopics(asList(topic1, topic2));
            }
        }
    }
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