private static void waitUntilMetadataIsPropagated(final List<KafkaServer> servers,
                                                     final String topic,
                                                     final int partition,
                                                     final long timeout) throws InterruptedException {
        final String baseReason = String.format("Metadata for topic=%s partition=%d was not propagated to all brokers within %d ms. ",
            topic, partition, timeout);

        retryOnExceptionWithTimeout(timeout, () -> {
            final List<KafkaServer> emptyPartitionInfos = new ArrayList<>();
            final List<KafkaServer> invalidBrokerIds = new ArrayList<>();

            for (final KafkaServer server : servers) {
                final MetadataCache metadataCache = server.dataPlaneRequestProcessor().metadataCache();
                final Option<UpdateMetadataPartitionState> partitionInfo =
                    metadataCache.getPartitionInfo(topic, partition);

                if (partitionInfo.isEmpty()) {
                    emptyPartitionInfos.add(server);
                    continue;
                }

                final UpdateMetadataPartitionState metadataPartitionState = partitionInfo.get();
                if (!FetchRequest.isValidBrokerId(metadataPartitionState.leader())) {
                    invalidBrokerIds.add(server);
                }
            }

            final String reason = baseReason + ". Brokers without partition info: " + emptyPartitionInfos +
                ". Brokers with invalid broker id for partition leader: " + invalidBrokerIds;
            assertThat(reason, emptyPartitionInfos.isEmpty() && invalidBrokerIds.isEmpty());
        });
    }
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