private void produceConsumeAndShutdown(ClusterInstance cluster,
                                           String topic,
                                           String group,
                                           int numConsumers,
                                           GroupProtocol groupProtocol) throws Exception {
        produceMessages(cluster, topic, 100);
        try (AutoCloseable consumerGroupCloseable =
                     consumerGroupClosable(cluster, numConsumers, topic, group, groupProtocol)) {
            awaitConsumerProgress(cluster, topic, group, 100);
        }
    }
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