private void awaitConsumerProgress(ClusterInstance cluster,
                                       String topic,
                                       String group,
                                       long count) throws Exception {
        try (Admin admin = Admin.create(singletonMap(BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers()))) {
            Supplier<Long> offsets = () -> {
                try {
                    return admin.listConsumerGroupOffsets(group)
                            .all().get().get(group)
                            .entrySet()
                            .stream()
                            .filter(e -> e.getKey().topic().equals(topic))
                            .mapToLong(e -> e.getValue().offset())
                            .sum();
                } catch (InterruptedException | ExecutionException e) {
                    throw new RuntimeException(e);
                }
            };
            TestUtils.waitForCondition(() -> offsets.get() == count,
                    "Expected that consumer group has consumed all messages from topic/partition. " +
                            "Expected offset: " + count +
                            ". Actual offset: " + offsets.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