public static void verifyKeyValueTimestamps(final Properties consumerConfig,
                                                final String topic,
                                                final Set<KeyValueTimestamp<String, Long>> expected) {
        final List<ConsumerRecord<String, Long>> results;
        try {
            results = waitUntilMinRecordsReceived(consumerConfig, topic, expected.size());
        } catch (final Exception e) {
            throw new RuntimeException(e);
        }

        if (results.size() != expected.size()) {
            throw new AssertionError(printRecords(results) + " != " + expected);
        }

        final Set<KeyValueTimestamp<String, Long>> actual =
            results.stream()
                   .map(result -> new KeyValueTimestamp<>(result.key(), result.value(), result.timestamp()))
                   .collect(Collectors.toSet());

        assertThat(actual, equalTo(expected));
    }
Waits for the given {@link KafkaStreams} instances to all be in a specific {@link State}.
Prefer {@link #startApplicationAndWaitUntilRunning(List, Duration)} when possible
because this method uses polling, which can be more error prone and slightly slower.

@param streamsList the list of streams instances to run.
@param state the expected state that all the streams to be in within timeout
@param timeout the time to wait for the streams to all be in the specific state.

@throws InterruptedException if the streams doesn't change to the expected state in time.
streams/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java