public static void startApplicationAndWaitUntilRunning(final List<KafkaStreams> streamsList,
                                                           final Duration timeout) throws Exception {
        final Lock stateLock = new ReentrantLock();
        final Condition stateUpdate = stateLock.newCondition();
        final Map<KafkaStreams, State> stateMap = new HashMap<>();
        for (final KafkaStreams streams : streamsList) {
            stateMap.put(streams, streams.state());
            final StateListener prevStateListener = getStateListener(streams);
            final StateListener newStateListener = (newState, oldState) -> {
                stateLock.lock();
                try {
                    stateMap.put(streams, newState);
                    if (newState == State.RUNNING) {
                        if (stateMap.values().stream().allMatch(state -> state == State.RUNNING)) {
                            stateUpdate.signalAll();
                        }
                    }
                } finally {
                    stateLock.unlock();
                }
            };

            streams.setStateListener(prevStateListener != null
                ? new CompositeStateListener(prevStateListener, newStateListener)
                : newStateListener);
        }

        for (final KafkaStreams streams : streamsList) {
            streams.start();
        }

        final long expectedEnd = System.currentTimeMillis() + timeout.toMillis();
        stateLock.lock();
        try {
            // We use while true here because we want to run this test at least once, even if the
            // timeout has expired
            while (true) {
                final Map<KafkaStreams, State> nonRunningStreams = new HashMap<>();
                for (final Entry<KafkaStreams, State> entry : stateMap.entrySet()) {
                    if (entry.getValue() != State.RUNNING) {
                        nonRunningStreams.put(entry.getKey(), entry.getValue());
                    }
                }

                if (nonRunningStreams.isEmpty()) {
                    return;
                }

                final long millisRemaining = expectedEnd - System.currentTimeMillis();
                if (millisRemaining <= 0) {
                    fail(
                        nonRunningStreams.size() + " out of " + streamsList.size() + " Streams clients did not reach the RUNNING state. " +
                            "Non-running Streams clients: " + nonRunningStreams
                    );
                }

                stateUpdate.await(millisRemaining, TimeUnit.MILLISECONDS);
            }
        } finally {
            stateLock.unlock();
        }
    }
Starts the given {@link KafkaStreams} instances and waits for all of them to reach the
{@link State#RUNNING} state at the same time. Note that states may change between the time
that this method returns and the calling function executes its next statement.<p>

If the application is already started, use {@link #waitForApplicationState(List, State, Duration)}
to wait for instances to reach {@link State#RUNNING} state.

@param streamsList the list of streams instances to run.
@param timeout the time to wait for the streams to all be in {@link State#RUNNING} state.
streams/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java