public static <K, V> List<KeyValueTimestamp<K, V>> waitUntilMinKeyValueWithTimestampRecordsReceived(final Properties consumerConfig,
                                                                                                        final String topic,
                                                                                                        final int expectedNumRecords,
                                                                                                        final long waitTime) throws Exception {
        final List<KeyValueTimestamp<K, V>> accumData = new ArrayList<>();
        final String reason = String.format(
            "Did not receive all %d records from topic %s within %d ms",
            expectedNumRecords,
            topic,
            waitTime
        );
        try (final Consumer<K, V> consumer = createConsumer(consumerConfig)) {
            retryOnExceptionWithTimeout(waitTime, () -> {
                final List<KeyValueTimestamp<K, V>> readData =
                    readKeyValuesWithTimestamp(topic, consumer, waitTime, expectedNumRecords);
                accumData.addAll(readData);
                assertThat(reason, accumData.size(), is(greaterThanOrEqualTo(expectedNumRecords)));
            });
        }
        return accumData;
    }
Wait until enough data (timestamped key-value records) has been consumed.

@param consumerConfig     Kafka Consumer configuration
@param topic              Kafka topic to consume from
@param expectedNumRecords Minimum number of expected records
@param waitTime           Upper bound of waiting time in milliseconds
@return All the records consumed, or null if no records are consumed
@param <K>                Key type of the data records
@param <V>                Value type of the data records
streams/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java