public static void waitForApplicationState(final List<KafkaStreams> streamsList,
                                               final State state,
                                               final Duration timeout) throws InterruptedException {
        retryOnExceptionWithTimeout(timeout.toMillis(), () -> {
            final Map<KafkaStreams, State> streamsToStates = streamsList
                .stream()
                .collect(Collectors.toMap(stream -> stream, KafkaStreams::state));

            final Map<KafkaStreams, State> wrongStateMap = streamsToStates.entrySet()
                .stream()
                .filter(entry -> entry.getValue() != state)
                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));

            final String reason = String.format(
                "Expected all streams instances in %s to be %s within %d ms, but the following were not: %s",
                streamsList,
                state,
                timeout.toMillis(),
                wrongStateMap
            );
            assertThat(reason, wrongStateMap.isEmpty());
        });
    }
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