public static <R> StateQueryResult<R> iqv2WaitForResult(
        final KafkaStreams kafkaStreams,
        final StateQueryRequest<R> request) {

        final long start = System.currentTimeMillis();
        final long deadline = start + DEFAULT_TIMEOUT;

        StateQueryResult<R> result;
        do {
            if (Thread.currentThread().isInterrupted()) {
                fail("Test was interrupted.");
            }

            result = kafkaStreams.query(request);
            final LinkedList<QueryResult<R>> allResults = getAllResults(result);

            if (allResults.isEmpty()) {
                sleep(100L);
            } else {
                final boolean needToWait = allResults
                    .stream()
                    .anyMatch(IntegrationTestUtils::needToWait);
                if (needToWait) {
                    sleep(100L);
                } else {
                    return result;
                }
            }
        } while (System.currentTimeMillis() < deadline);

        throw new TimeoutException(
            "The query never returned within the bound. Last result: "
            + result
        );
    }
Repeatedly runs the query until the response is valid and then return the response.
<p>
Validity in this case means that the response position is up to the specified bound.
<p>
Once position bounding is generally supported, we should migrate tests to wait on the
expected response position.
streams/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java