This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 9c8b2901d3e Require consecutive readiness checks in the
integration-test base (#20015)
9c8b2901d3e is described below
commit 9c8b2901d3eb3788115d62e8face820874d6743d
Author: Xiang Fu <[email protected]>
AuthorDate: Mon Oct 26 17:45:19 2026 -0700
Require consecutive readiness checks in the integration-test base (#20015)
Require _serverStarters.size() consecutive successful document-count checks
in BaseClusterIntegrationTest.waitForAllDocsLoaded. Reset the streak on
count
mismatches or query exceptions, retaining the polling interval and overall
timeout. Keep the change limited to the integration-test parent class.
How to reproduce:
Run WindowFunnelTest while the two-server cluster loads segments. A
successful
count can come from a ready replica while another server is still loading,
allowing a subsequent multi-stage funnel query to return 36 groups instead
of 40.
---
.../tests/BaseClusterIntegrationTest.java | 20 ++++++++++++++++++--
1 file changed, 18 insertions(+), 2 deletions(-)
diff --git
a/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTest.java
b/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTest.java
index 327eae47fd0..e3dfbaf19a0 100644
---
a/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTest.java
+++
b/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTest.java
@@ -38,6 +38,7 @@ import java.util.Properties;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import javax.annotation.Nullable;
import org.apache.commons.io.FileUtils;
@@ -1021,10 +1022,25 @@ public abstract class BaseClusterIntegrationTest
extends ClusterTest {
waitForAllDocsLoaded(getTableName(), timeoutMs);
}
+ /// Requires as many consecutive successful document count checks as there
are servers in the cluster,
+ /// since servers can finish loading their segments at different times.
protected void waitForAllDocsLoaded(String tableName, long timeoutMs) {
long countStarResult = getCountStarResult();
- TestUtils.waitForCondition(() -> getCurrentCountStarResult(tableName) ==
countStarResult, 100L, timeoutMs,
- "Failed to load " + countStarResult + " documents",
Duration.ofMillis(timeoutMs / 10));
+ int numServers = _serverStarters.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;
+ }
+ }, 100L, timeoutMs, "Failed to load " + countStarResult + " documents on
all servers",
+ Duration.ofMillis(timeoutMs / 10));
}
protected void waitForAnyDocLoaded(String tableName, long timeoutMs) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
