vvcephei commented on a change in pull request #11581: URL: https://github.com/apache/kafka/pull/11581#discussion_r766226822
########## File path: streams/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java ########## @@ -125,22 +128,81 @@ final long deadline = start + DEFAULT_TIMEOUT; do { + if (Thread.currentThread().isInterrupted()) { + fail("Test was interrupted."); + } final StateQueryResult<R> result = kafkaStreams.query(request); - if (result.getPartitionResults().keySet().containsAll(partitions) - || result.getGlobalResult() != null) { + if (result.getPartitionResults().keySet().containsAll(partitions)) { return result; } else { - try { - Thread.sleep(100L); - } catch (final InterruptedException e) { - throw new RuntimeException(e); - } + sleep(100L); } } while (System.currentTimeMillis() < deadline); throw new TimeoutException("The query never returned the desired partitions"); } + /** + * 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. + */ + 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()) { Review comment: You'll only get a `NOT_PRESENT` response if you specifically request a partition. The default is to just get all locally present partitions. This check is actually just an assumption that in the context of an integration test, if you call this method, you're probably expecting at least one result. It is good to note, though, that if a test is looking for results for a specific set of partitions, it should include that in the query. ########## File path: streams/src/main/java/org/apache/kafka/streams/state/internals/StoreQueryUtils.java ########## @@ -62,4 +71,32 @@ public static void updatePosition( position.withComponent(meta.topic(), meta.partition(), meta.offset()); } } + + public static boolean isPermitted( + final Position position, + final PositionBound positionBound, + final int partition + ) { + if (positionBound.isUnbounded()) { + return true; + } else { + final Position bound = positionBound.position(); + for (final String topic : bound.getTopics()) { + final Map<Integer, Long> partitionBounds = bound.getBound(topic); + final Map<Integer, Long> seenPartitionBounds = position.getBound(topic); Review comment: Woah, you're absolutely right. I'll fix it. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org