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 26556ffc062 test: Make `EmbeddedKafkaSupervisorTest` wait for Broker
queryability (#20620)
26556ffc062 is described below
commit 26556ffc0629fa62ec89c52e0e9afe09e7b93d71
Author: Andreas Maechler <[email protected]>
AuthorDate: Fri Oct 16 23:08:28 2026 -0600
test: Make `EmbeddedKafkaSupervisorTest` wait for Broker queryability
(#20620)
* Fix flaky
EmbeddedKafkaSupervisorTest.testSupervisorIngestsOnlySelectedPartitions
waitForSelectedRows only waited for the indexer's ingest/events/processed
metric, so the SQL assertions could run before the Broker had the datasource
in its schema and fail with "Object 'datasource_xxx' not found".
Wait for ingest/handoff/count instead, then for the segments to be queryable
through the Broker with waitForAllSegmentsToBeQueryable. Handoff follows
publish, which commits the partition 1 offset before the supervisor spec is
replaced, so the replacement supervisor resumes from it.
Submit both specs through submitSupervisor so the shared teardown suspends
the supervisor and cancels its tasks even when the test fails.
* Wait for Broker queryability in the other EmbeddedKafkaSupervisorTest SQL
checks
test_runKafkaSupervisor checked the row count after an indexer-only
ingest/events/processed wait, and header filtering queried right after
handoff without checking the Broker.
Add EmbeddedClusterApis.waitForSegmentsToBeQueryable, which takes the
segment count so it also covers unpublished realtime segments, and make
waitForAllSegmentsToBeQueryable delegate to it. test_runKafkaSupervisor
waits for its 10 segments with it. Header filtering and the
selected-partitions test share a waitForHandedOffRows helper.
---
.../simulate/EmbeddedKafkaSupervisorTest.java | 32 ++++++++++------------
.../testing/embedded/EmbeddedClusterApis.java | 12 +++++++-
2 files changed, 25 insertions(+), 19 deletions(-)
diff --git
a/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/simulate/EmbeddedKafkaSupervisorTest.java
b/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/simulate/EmbeddedKafkaSupervisorTest.java
index f407b13fa98..762de1e983d 100644
---
a/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/simulate/EmbeddedKafkaSupervisorTest.java
+++
b/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/simulate/EmbeddedKafkaSupervisorTest.java
@@ -121,11 +121,11 @@ public class EmbeddedKafkaSupervisorTest extends
EmbeddedClusterTestBase
final KafkaSupervisorSpec spec = newKafkaSupervisor()
.withIoConfig(io ->
io.withPartitionIds(Set.of(1)).withTaskCount(2).withStopTaskCount(1))
.build(dataSource, topic);
- cluster.callApi().postSupervisor(spec);
+ submitSupervisor(spec);
// One record per partition, but only partition 1 is selected.
kafkaServer.produceRecordsToTopic(partitionRecords(topic, 4));
- waitForSelectedRows(dataSource, 1);
+ waitForHandedOffRows(1);
Assertions.assertEquals("1", cluster.runSql("SELECT COUNT(*) FROM %s",
dataSource));
Assertions.assertEquals("0", cluster.runSql("SELECT COUNT(*) FROM %s WHERE
item != 'p1'", dataSource));
@@ -134,12 +134,11 @@ public class EmbeddedKafkaSupervisorTest extends
EmbeddedClusterTestBase
final KafkaSupervisorSpec expanded = newKafkaSupervisor()
.withIoConfig(io -> io.withPartitionIds(Set.of(1,
3)).withTaskCount(2).withStopTaskCount(1))
.build(dataSource, topic);
- cluster.callApi().postSupervisor(expanded);
- waitForSelectedRows(dataSource, 2);
+ submitSupervisor(expanded);
+ waitForHandedOffRows(2);
Assertions.assertEquals("2", cluster.runSql("SELECT COUNT(*) FROM %s",
dataSource));
Assertions.assertEquals("0", cluster.runSql("SELECT COUNT(*) FROM %s WHERE
item = 'p2'", dataSource));
- cluster.callApi().postSupervisor(expanded.createSuspendedSpec());
}
private static List<ProducerRecord<byte[], byte[]>> partitionRecords(String
topic, int partitions)
@@ -156,12 +155,17 @@ public class EmbeddedKafkaSupervisorTest extends
EmbeddedClusterTestBase
return records;
}
- private void waitForSelectedRows(String source, int count)
+ /**
+ * Waits for {@code count} segments, one per row with maxRowsPerSegment = 1,
to be handed off and queryable through
+ * the Broker. Handoff follows publish, so the task's offsets are committed
by the time this returns.
+ */
+ private void waitForHandedOffRows(int count)
{
indexer.latchableEmitter().waitForEventAggregate(
- event ->
event.hasMetricName("ingest/events/processed").hasDimension(DruidMetrics.DATASOURCE,
source),
+ event ->
event.hasMetricName("ingest/handoff/count").hasDimension(DruidMetrics.DATASOURCE,
dataSource),
agg -> agg.hasSumAtLeast(count)
);
+ cluster.callApi().waitForAllSegmentsToBeQueryable(dataSource, coordinator,
30_000L);
}
@Test
@@ -202,13 +206,9 @@ public class EmbeddedKafkaSupervisorTest extends
EmbeddedClusterTestBase
Assertions.assertEquals(TaskState.RUNNING, task.getStatusCode());
}
- // Wait until all produced records have been ingested before verifying the
row count,
+ // Wait until the Broker can query all produced records before verifying
the row count,
// otherwise the query below can race ingestion and observe fewer than the
expected rows
- indexer.latchableEmitter().waitForEventAggregate(
- event -> event.hasMetricName("ingest/events/processed")
- .hasDimension(DruidMetrics.DATASOURCE, dataSource),
- agg -> agg.hasSumAtLeast(expectedSegments)
- );
+ cluster.callApi().waitForSegmentsToBeQueryable(dataSource,
expectedSegments, 30_000L);
// Verify the count of rows ingested into the datasource so far
Assertions.assertEquals(
@@ -270,11 +270,7 @@ public class EmbeddedKafkaSupervisorTest extends
EmbeddedClusterTestBase
// Wait for segment handoff without suspending the supervisor; its short
task duration publishes the segments.
// Suspending while ingestion is in progress can replace the supervisor
while the task's checkpoint request is in
// flight, which fails the task before it publishes anything.
- indexer.latchableEmitter().waitForEventAggregate(
- event -> event.hasMetricName("ingest/handoff/count")
- .hasDimension(DruidMetrics.DATASOURCE, dataSource),
- agg -> agg.hasSumAtLeast(expectedRecords)
- );
+ waitForHandedOffRows(expectedRecords);
// Verify only filtered records were ingested
Assertions.assertEquals(String.valueOf(expectedRecords),
cluster.runSql("SELECT COUNT(*) FROM %s", dataSource));
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 a4fb736ee64..b25c6567ad9 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
@@ -372,7 +372,6 @@ public class EmbeddedClusterApis implements EmbeddedResource
long timeoutMillis
)
{
- final String escapedDataSource = StringUtils.escapeSql(dataSource);
final int numSegments = (int) coordinator
.bindings()
.segmentsMetadataStorage()
@@ -381,6 +380,17 @@ public class EmbeddedClusterApis implements
EmbeddedResource
.filter(segment -> !segment.isTombstone())
.count();
+ waitForSegmentsToBeQueryable(dataSource, numSegments, timeoutMillis);
+ }
+
+ /**
+ * Like {@link #waitForAllSegmentsToBeQueryable}, but waits for a given
number of segments, which may include
+ * unpublished realtime segments that are not in the metadata store yet.
+ */
+ public void waitForSegmentsToBeQueryable(String dataSource, int numSegments,
long timeoutMillis)
+ {
+ final String escapedDataSource = StringUtils.escapeSql(dataSource);
+
waitForResult(
() -> runSql(
"SELECT COUNT(*) FROM sys.segments WHERE datasource='%s' AND
is_available = 1",
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
