private void resetAndAssertOffsets(ClusterInstance cluster,
                                       String[] args,
                                       long expectedOffset,
                                       boolean dryRun,
                                       List<String> topics) {
        try (ConsumerGroupCommand.ConsumerGroupService service = getConsumerGroupService(args)) {
            Map<String, Map<TopicPartition, Long>> topicToExpectedOffsets = getTopicExceptOffsets(topics, expectedOffset);
            Map<String, Map<TopicPartition, OffsetAndMetadata>> resetOffsetsResultByGroup =
                    resetOffsets(service);
            for (final String topic : topics) {
                resetOffsetsResultByGroup.forEach((group, partitionInfo) -> {
                    Map<TopicPartition, Long> priorOffsets = committedOffsets(cluster, topic, group);
                    assertEquals(topicToExpectedOffsets.get(topic), partitionToOffsets(topic, partitionInfo));
                    assertEquals(dryRun ? priorOffsets : topicToExpectedOffsets.get(topic),
                            committedOffsets(cluster, topic, group));
                });
            }
        }
    }
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