private static <K, V> List<ConsumerRecord<K, V>> readRecords(final String topic,
                                                                 final Consumer<K, V> consumer,
                                                                 final long waitTime,
                                                                 final int maxMessages) {
        final List<ConsumerRecord<K, V>> consumerRecords;
        consumer.subscribe(singletonList(topic));
        final int pollIntervalMs = 100;
        consumerRecords = new ArrayList<>();
        int totalPollTimeMs = 0;
        while (totalPollTimeMs < waitTime &&
            continueConsuming(consumerRecords.size(), maxMessages)) {
            totalPollTimeMs += pollIntervalMs;
            final ConsumerRecords<K, V> records = consumer.poll(Duration.ofMillis(pollIntervalMs));

            for (final ConsumerRecord<K, V> record : records) {
                consumerRecords.add(record);
            }
        }
        return consumerRecords;
    }
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