private Map<TopicPartition, Long> committedOffsets(ClusterInstance cluster,
                                                       String topic,
                                                       String group) {
        try (Admin admin = Admin.create(singletonMap(BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers()))) {
            return admin.listConsumerGroupOffsets(group)
                    .all().get()
                    .get(group).entrySet()
                    .stream()
                    .filter(e -> e.getKey().topic().equals(topic))
                    .collect(toMap(Map.Entry::getKey, e -> e.getValue().offset()));
        } catch (ExecutionException | InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
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