xiangfu0 commented on code in PR #20015:
URL: https://github.com/apache/pinot/pull/20015#discussion_r4383460226
##########
pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/CustomDataQueryClusterIntegrationTest.java:
##########
@@ -118,6 +119,28 @@ public void setUp()
LOGGER.warn("Finished setting up integration test class: {}",
getClass().getSimpleName());
}
+ /// Requires one consecutive successful document count check per server,
since some servers can finish loading
+ /// their segments before others.
+ @Override
+ protected void waitForAllDocsLoaded(String tableName, long timeoutMs) {
Review Comment:
This replaces the deterministic `waitForNoOptionalSegments` barrier from the
first commit. Since single-stage count(*) already includes optional segments, N
consecutive matching counts don't directly prove the broker has stopped routing
a segment as optional — the thing that made the multi-stage funnel query miss a
segment. If your harness shows the consecutive-count check reliably covers that
case too, great; just flagging that it's a probabilistic guarantee where the
barrier was exact.
##########
pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/CustomDataQueryClusterIntegrationTest.java:
##########
@@ -118,6 +119,28 @@ public void setUp()
LOGGER.warn("Finished setting up integration test class: {}",
getClass().getSimpleName());
}
+ /// Requires one consecutive successful document count check per server,
since some servers can finish loading
+ /// their segments before others.
+ @Override
+ protected void waitForAllDocsLoaded(String tableName, long timeoutMs) {
+ long countStarResult = getCountStarResult();
+ int numServers = getSharedServerStarters().size();
+ AtomicInteger consecutiveSuccesses = new AtomicInteger();
+ TestUtils.waitForCondition(() -> {
+ try {
+ if (getCurrentCountStarResult(tableName) != countStarResult) {
+ consecutiveSuccesses.set(0);
+ return false;
+ }
+ return consecutiveSuccesses.incrementAndGet() >= numServers;
+ } catch (Exception e) {
+ consecutiveSuccesses.set(0);
+ throw e;
Review Comment:
Minor: this `throw e` doesn't actually abort the wait. The
`waitForCondition(SupplierWithException, ..., Duration)` overload catches the
exception, logs it, and keeps polling — so the real effect here is just the
streak reset on the line above, and the throw reads as if it propagates when it
doesn't. Dropping it (or a one-line comment) would avoid the confusion.
##########
pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/CustomDataQueryClusterIntegrationTest.java:
##########
@@ -118,6 +119,28 @@ public void setUp()
LOGGER.warn("Finished setting up integration test class: {}",
getClass().getSimpleName());
}
+ /// Requires one consecutive successful document count check per server,
since some servers can finish loading
+ /// their segments before others.
+ @Override
+ protected void waitForAllDocsLoaded(String tableName, long timeoutMs) {
Review Comment:
This replaces the deterministic `waitForNoOptionalSegments` barrier from the
first commit. Since single-stage count(*) already includes optional segments, N
consecutive matching counts don't directly prove the broker has stopped routing
a segment as optional — the thing that made the multi-stage funnel query miss a
segment. If your harness shows the consecutive-count check reliably covers that
case too, great; just flagging that it's a probabilistic guarantee where the
barrier was exact.
##########
pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/CustomDataQueryClusterIntegrationTest.java:
##########
@@ -118,6 +119,28 @@ public void setUp()
LOGGER.warn("Finished setting up integration test class: {}",
getClass().getSimpleName());
}
+ /// Requires one consecutive successful document count check per server,
since some servers can finish loading
+ /// their segments before others.
+ @Override
+ protected void waitForAllDocsLoaded(String tableName, long timeoutMs) {
+ long countStarResult = getCountStarResult();
+ int numServers = getSharedServerStarters().size();
+ AtomicInteger consecutiveSuccesses = new AtomicInteger();
+ TestUtils.waitForCondition(() -> {
+ try {
+ if (getCurrentCountStarResult(tableName) != countStarResult) {
+ consecutiveSuccesses.set(0);
+ return false;
+ }
+ return consecutiveSuccesses.incrementAndGet() >= numServers;
+ } catch (Exception e) {
+ consecutiveSuccesses.set(0);
+ throw e;
Review Comment:
Minor: this `throw e` doesn't actually abort the wait. The
`waitForCondition(SupplierWithException, ..., Duration)` overload catches the
exception, logs it, and keeps polling — so the real effect here is just the
streak reset on the line above, and the throw reads as if it propagates when it
doesn't. Dropping it (or a one-line comment) would avoid the confusion.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
