This is an automated email from the ASF dual-hosted git repository.
FrankChen021 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new b7b560dfd29 test: poll Broker state instead of schema refresh metrics
in QueryTestBase (#20319)
b7b560dfd29 is described below
commit b7b560dfd29ac1fbeae75a35c9026bacef549bdf
Author: Frank Chen <[email protected]>
AuthorDate: Thu Sep 24 10:03:56 2026 +0800
test: poll Broker state instead of schema refresh metrics in QueryTestBase
(#20319)
* test: poll Broker state instead of schema refresh metrics in QueryTestBase
QueryLaningTest and other QueryTestBase subclasses intermittently timed out
in
waitForAllSegmentsToBeAvailable, which waits for an aggregate count of
Broker
schema refresh metrics. Add waitForAllSegmentsToBeQueryable, which polls the
state that SQL queries actually observe: available segments in sys.segments
and the datasource being present in INFORMATION_SCHEMA.TABLES, and use it in
QueryTestBase.ingestBasicData().
* Treat empty or non-numeric sys.segments poll results as not yet ready
A NumberFormatException inside the ResultWaiter supplier/matcher would
abort the
wait instead of retrying.
* test: escape datasource in Broker polling queries
---
.../testing/embedded/query/QueryTestBase.java | 4 +-
.../testing/embedded/EmbeddedClusterApis.java | 57 ++++++++++++++++++++++
2 files changed, 60 insertions(+), 1 deletion(-)
diff --git
a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/query/QueryTestBase.java
b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/query/QueryTestBase.java
index e19850276e2..686d7dc253c 100644
---
a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/query/QueryTestBase.java
+++
b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/query/QueryTestBase.java
@@ -128,7 +128,9 @@ public abstract class QueryTestBase extends
EmbeddedClusterTestBase
final IndexTask task =
MoreResources.Task.BASIC_INDEX.get().dataSource(datasourceName).withId(taskId);
cluster.callApi().onLeaderOverlord(o -> o.runTask(taskId, task));
cluster.callApi().waitForTaskToSucceed(taskId, overlord);
- cluster.callApi().waitForAllSegmentsToBeAvailable(datasourceName,
coordinator, broker);
+ // Poll the state observed by SQL queries rather than waiting for Broker
schema refresh metrics,
+ // which has been observed to time out intermittently in CI.
+ cluster.callApi().waitForAllSegmentsToBeQueryable(datasourceName,
coordinator, 100_000L);
return datasourceName;
}
diff --git
a/services/src/test/java/org/apache/druid/testing/embedded/EmbeddedClusterApis.java
b/services/src/test/java/org/apache/druid/testing/embedded/EmbeddedClusterApis.java
index 6ae8750b8d8..8082bced562 100644
---
a/services/src/test/java/org/apache/druid/testing/embedded/EmbeddedClusterApis.java
+++
b/services/src/test/java/org/apache/druid/testing/embedded/EmbeddedClusterApis.java
@@ -332,6 +332,63 @@ public class EmbeddedClusterApis implements
EmbeddedResource
);
}
+ /**
+ * Waits for all non-tombstone used segments of the given datasource to be
+ * reported as available in {@code sys.segments} and for the datasource to be
+ * present in the Broker SQL schema, by polling the Broker.
+ * <p>
+ * Unlike {@link #waitForAllSegmentsToBeAvailable}, this method does not
depend
+ * on schema refresh metrics being emitted by the Broker and verifies the
state
+ * that SQL queries actually observe.
+ *
+ * @param timeoutMillis maximum time to wait
+ */
+ public void waitForAllSegmentsToBeQueryable(
+ String dataSource,
+ EmbeddedCoordinator coordinator,
+ long timeoutMillis
+ )
+ {
+ final String escapedDataSource = StringUtils.escapeSql(dataSource);
+ final int numSegments = (int) coordinator
+ .bindings()
+ .segmentsMetadataStorage()
+ .retrieveAllUsedSegments(dataSource, Segments.INCLUDING_OVERSHADOWED)
+ .stream()
+ .filter(segment -> !segment.isTombstone())
+ .count();
+
+ waitForResult(
+ () -> runSql(
+ "SELECT COUNT(*) FROM sys.segments WHERE datasource='%s' AND
is_available = 1",
+ escapedDataSource
+ ),
+ result -> parseCountOrZero(result) >= numSegments
+ ).withTimeoutMillis(timeoutMillis).go();
+
+ waitForResult(
+ () -> runSql(
+ "SELECT COUNT(*) FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA
= 'druid' AND TABLE_NAME = '%s'",
+ escapedDataSource
+ ),
+ result -> "1".equals(result.trim())
+ ).withTimeoutMillis(timeoutMillis).go();
+ }
+
+ /**
+ * Parses a {@code COUNT(*)} result, treating an empty or non-numeric result
+ * (e.g. while the Broker is still initializing) as zero so that polling
continues.
+ */
+ private static long parseCountOrZero(String result)
+ {
+ try {
+ return Long.parseLong(result.trim());
+ }
+ catch (NumberFormatException e) {
+ return 0L;
+ }
+ }
+
/**
* Waits for all used segments (including overshadowed) of the given
datasource
* to be queryable by Brokers when centralized schema is enabled.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
