From 9f6d3536aea61f37dc9e93393d6e7d032c0a962c Mon Sep 17 00:00:00 2001 From: Frank Chen Date: Thu, 10 Sep 2026 23:26:41 +0800 Subject: [PATCH 1/3] test: poll sys.segments in verifyNumVisibleSegmentsIs to fix flaky compaction tests CompactionTaskTest intermittently failed with an empty sys.segments count after waiting for a single segment/metadataCache/sync/time event, because an in-flight sync can still observe the previous state. Keep the immediate Overlord-side assertion and poll the Broker-side sys.segments count with a deadline, reporting the last observed result on failure. --- .../embedded/compact/CompactionTestBase.java | 4 +-- .../testing/embedded/EmbeddedClusterApis.java | 31 +++++++++++++------ 2 files changed, 24 insertions(+), 11 deletions(-) diff --git a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/compact/CompactionTestBase.java b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/compact/CompactionTestBase.java index b714cd00859e..aefb9770bab4 100644 --- a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/compact/CompactionTestBase.java +++ b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/compact/CompactionTestBase.java @@ -97,8 +97,8 @@ protected List getSegmentIntervals() protected void verifySegmentsCount(int numExpectedSegments) { - // Ensure that Broker has synced latest segments from the Coordinator - broker.latchableEmitter().waitForNextEvent(event -> event.hasMetricName("segment/metadataCache/sync/time")); + // The Overlord state is verified immediately; the Broker-side sys.segments view is polled + // until it matches, since a single metadata cache sync may still observe the previous state. cluster.callApi().verifyNumVisibleSegmentsIs(numExpectedSegments, dataSource, overlord); } 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 6ae8750b8d8e..93c9f361ca1c 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 @@ -287,15 +287,28 @@ public void verifyNumVisibleSegmentsIs(int numExpectedSegments, String dataSourc segmentCount, "Segment count mismatch" ); - Assertions.assertEquals( - String.valueOf(segmentCount), - runSql( - "SELECT COUNT(*) FROM sys.segments WHERE datasource='%s'" - + " AND is_overshadowed = 0 AND is_available = 1", - dataSource - ), - "Segment count mismatch in sys.segments table" - ); + + // The Broker learns about segment changes asynchronously from the Coordinator, so the + // sys.segments table may briefly lag behind the metadata store. Poll instead of asserting once. + final String expectedCount = String.valueOf(segmentCount); + final String sql = "SELECT COUNT(*) FROM sys.segments WHERE datasource='%s'" + + " AND is_overshadowed = 0 AND is_available = 1"; + try { + waitForResult(() -> runSql(sql, dataSource), expectedCount::equals) + .withTimeoutMillis(60_000) + .go(); + } + catch (ISE e) { + throw new AssertionError( + StringUtils.format( + "Segment count mismatch in sys.segments table: expected[%s] segments for datasource[%s]. %s", + expectedCount, + dataSource, + e.getMessage() + ), + e + ); + } } /** From bf951bfbe74049023ee85f8f4a1f2545fe41057b Mon Sep 17 00:00:00 2001 From: Frank Chen Date: Sat, 19 Sep 2026 22:35:49 +0800 Subject: [PATCH 2/3] test: wait for current segment IDs after compaction --- .../testing/embedded/EmbeddedClusterApis.java | 16 ++++++++++++---- 1 file changed, 12 insertions(+), 4 deletions(-) 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 93c9f361ca1c..2e372aff31a4 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 @@ -61,6 +61,7 @@ import java.util.TreeSet; import java.util.function.Function; import java.util.function.Predicate; +import java.util.stream.Collectors; /** * Contains various utility methods to interact with an {@link EmbeddedDruidCluster}. @@ -281,7 +282,8 @@ public List getSortedSegmentIntervals(String dataSource, EmbeddedOverl */ public void verifyNumVisibleSegmentsIs(int numExpectedSegments, String dataSource, EmbeddedOverlord overlord) { - int segmentCount = getVisibleUsedSegments(dataSource, overlord).size(); + final Set visibleSegments = getVisibleUsedSegments(dataSource, overlord); + final int segmentCount = visibleSegments.size(); Assertions.assertEquals( numExpectedSegments, segmentCount, @@ -289,12 +291,18 @@ public void verifyNumVisibleSegmentsIs(int numExpectedSegments, String dataSourc ); // The Broker learns about segment changes asynchronously from the Coordinator, so the - // sys.segments table may briefly lag behind the metadata store. Poll instead of asserting once. + // sys.segments table may briefly lag behind the metadata store. Match the segment IDs as well + // as the count because compaction can replace segments without changing their number. final String expectedCount = String.valueOf(segmentCount); + final String expectedSegmentIds = visibleSegments + .stream() + .map(segment -> StringUtils.format("'%s'", StringUtils.escapeSql(segment.getId().toString()))) + .collect(Collectors.joining(", ")); final String sql = "SELECT COUNT(*) FROM sys.segments WHERE datasource='%s'" - + " AND is_overshadowed = 0 AND is_available = 1"; + + " AND is_overshadowed = 0 AND is_available = 1" + + (expectedSegmentIds.isEmpty() ? "" : " AND segment_id IN (" + expectedSegmentIds + ")"); try { - waitForResult(() -> runSql(sql, dataSource), expectedCount::equals) + waitForResult(() -> runSql(sql, StringUtils.escapeSql(dataSource)), expectedCount::equals) .withTimeoutMillis(60_000) .go(); } From 91e4fec77478ab256d261167e0d3ebb9989766b5 Mon Sep 17 00:00:00 2001 From: Frank Chen Date: Tue, 22 Sep 2026 18:07:24 +0800 Subject: [PATCH 3/3] test: preserve percent signs in segment ID filter --- .../druid/testing/embedded/EmbeddedClusterApis.java | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) 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 2e372aff31a4..091c86c1da9c 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 @@ -298,11 +298,13 @@ public void verifyNumVisibleSegmentsIs(int numExpectedSegments, String dataSourc .stream() .map(segment -> StringUtils.format("'%s'", StringUtils.escapeSql(segment.getId().toString()))) .collect(Collectors.joining(", ")); + final String segmentIdFilter = expectedSegmentIds.isEmpty() + ? "" + : " AND segment_id IN (" + expectedSegmentIds + ")"; final String sql = "SELECT COUNT(*) FROM sys.segments WHERE datasource='%s'" - + " AND is_overshadowed = 0 AND is_available = 1" - + (expectedSegmentIds.isEmpty() ? "" : " AND segment_id IN (" + expectedSegmentIds + ")"); + + " AND is_overshadowed = 0 AND is_available = 1%s"; try { - waitForResult(() -> runSql(sql, StringUtils.escapeSql(dataSource)), expectedCount::equals) + waitForResult(() -> runSql(sql, StringUtils.escapeSql(dataSource), segmentIdFilter), expectedCount::equals) .withTimeoutMillis(60_000) .go(); }