@SuppressWarnings("WeakerAccess")
    public static void waitForTopicPartitions(final List<KafkaServer> servers,
                                              final List<TopicPartition> partitions,
                                              final long timeout) throws InterruptedException {
        final long end = System.currentTimeMillis() + timeout;
        for (final TopicPartition partition : partitions) {
            final long remaining = end - System.currentTimeMillis();
            if (remaining <= 0) {
                throw new AssertionError("timed out while waiting for partitions to become available. Timeout=" + timeout);
            }
            waitUntilMetadataIsPropagated(servers, partition.topic(), partition.partition(), remaining);
        }
    }
Wait until enough data (value records) has been consumed.

@param consumerConfig     Kafka Consumer configuration
@param topic              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
@throws AssertionError    if the given wait time elapses
streams/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java