private static <K, V> List<KeyValue<K, V>> readKeyValues(final String topic,
                                                             final Consumer<K, V> consumer,
                                                             final long waitTime,
                                                             final int maxMessages) {
        final List<KeyValue<K, V>> consumedValues = new ArrayList<>();
        final List<ConsumerRecord<K, V>> records = readRecords(topic, consumer, waitTime, maxMessages);
        for (final ConsumerRecord<K, V> record : records) {
            consumedValues.add(new KeyValue<>(record.key(), record.value()));
        }
        return consumedValues;
    }
Returns up to `maxMessages` by reading via the provided consumer (the topic(s) to read from
are already configured in the consumer).

@param topic          Kafka topic to read messages from
@param consumer       Kafka consumer
@param waitTime       Maximum wait time in milliseconds
@param maxMessages    Maximum number of messages to read via the consumer
@return The KeyValue elements retrieved via the consumer
streams/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java