@ClusterTemplate("generator")
    public void testResetOffsetsToLocalDateTime(ClusterInstance cluster) throws Exception {
        for (GroupProtocol groupProtocol : cluster.supportedGroupProtocols()) {
            String group = generateRandomGroupId();
            String topic = generateRandomTopic();

            DateTimeFormatter format = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSS");
            LocalDateTime dateTime = now().minusDays(1);
            String[] args = buildArgsForGroup(cluster, group,
                "--all-topics", "--to-datetime",
                format.format(dateTime), "--execute");

            produceMessages(cluster, topic, 100);

            try (AutoCloseable consumerGroupCloseable =
                         consumerGroupClosable(cluster, 1, topic, group, groupProtocol)) {
                awaitConsumerProgress(cluster, topic, group, 100L);
            }

            resetAndAssertOffsets(cluster, topic, args, 0);
        }
    }
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