You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by xi...@apache.org on 2026/10/27 00:45:25 UTC
(pinot) branch master updated: Require consecutive readiness checks in the integration-test base (#20015)
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 <xi...@apache.org>
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: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org