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 0387ffa9028 fix: stabilize task queue and ingestion CI tests (#20291)
0387ffa9028 is described below
commit 0387ffa90287033948275b9d3a727b0c27258cc9
Author: Frank Chen <[email protected]>
AuthorDate: Sat Sep 26 12:00:39 2026 +0800
fix: stabilize task queue and ingestion CI tests (#20291)
* fix: stabilize task queue and ingestion CI tests
* test: drain ingestion smoke test tasks during cleanup
* fix: compile supervisor autoscaling tests
* fix: clear stale datasource rebuild state on removal
* fix: satisfy guarded datasource rebuild cleanup
* fix: emit dataSource/removed metric at most once per table removal
The last-segment callback and an in-flight refresh can both observe the
same datasource removal. Gate the metric on the tables.remove() result in
both paths so exactly one segment/schemaCache/dataSource/removed event is
emitted, regardless of which path removes the table first.
* fix: guard the empty-signature removal metric as well
buildDataSourceRowSignature returns an empty (non-null) signature when a
datasource only has unrefreshed segments. That branch of refresh() still
emitted dataSource/removed unconditionally, so a last-segment callback
racing an in-flight refresh could report the removal twice. Apply the same
tables.remove() guard, and add a latch-controlled test for the interleaving
plus one for a never-initialised datasource.
---
.../embedded/indexing/IngestionSmokeTest.java | 54 ++++++-
.../indexing/KafkaBoundedSupervisorTest.java | 25 +++-
.../indexing/overlord/TaskQueueScaleTest.java | 20 +--
.../metadata/AbstractSegmentMetadataCache.java | 21 ++-
.../calcite/schema/BrokerSegmentMetadataCache.java | 37 +++--
.../schema/BrokerSegmentMetadataCacheTest.java | 165 +++++++++++++++++++++
6 files changed, 296 insertions(+), 26 deletions(-)
diff --git
a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/IngestionSmokeTest.java
b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/IngestionSmokeTest.java
index 2d22e80960b..b05c4731fb6 100644
---
a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/IngestionSmokeTest.java
+++
b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/IngestionSmokeTest.java
@@ -24,6 +24,7 @@ import org.apache.commons.io.IOUtils;
import org.apache.druid.common.utils.IdUtils;
import org.apache.druid.data.input.impl.CsvInputFormat;
import org.apache.druid.data.input.impl.TimestampSpec;
+import org.apache.druid.indexer.TaskStatusPlus;
import org.apache.druid.indexing.common.task.CompactionTask;
import org.apache.druid.indexing.common.task.IndexTask;
import org.apache.druid.indexing.common.task.NoopTask;
@@ -33,6 +34,9 @@ import org.apache.druid.indexing.kafka.KafkaIndexTaskModule;
import org.apache.druid.indexing.kafka.simulate.KafkaResource;
import org.apache.druid.indexing.kafka.supervisor.KafkaSupervisorSpec;
import org.apache.druid.indexing.overlord.Segments;
+import org.apache.druid.indexing.overlord.TaskMaster;
+import org.apache.druid.indexing.overlord.TaskRunner;
+import org.apache.druid.indexing.overlord.TaskRunnerWorkItem;
import org.apache.druid.indexing.overlord.supervisor.SupervisorStatus;
import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.java.util.common.Intervals;
@@ -40,6 +44,7 @@ import org.apache.druid.java.util.common.StringUtils;
import
org.apache.druid.metadata.storage.postgresql.PostgreSQLMetadataStorageModule;
import org.apache.druid.query.DruidMetrics;
import org.apache.druid.query.http.SqlTaskStatus;
+import org.apache.druid.rpc.indexing.OverlordClient;
import org.apache.druid.segment.metadata.Metric;
import org.apache.druid.tasklogs.TaskLogStreamer;
import org.apache.druid.testing.embedded.EmbeddedBroker;
@@ -67,6 +72,7 @@ import org.junit.jupiter.api.Test;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -145,7 +151,51 @@ public class IngestionSmokeTest extends
EmbeddedClusterTestBase
@AfterEach
public void cleanUp()
{
- markSegmentsAsUnused(dataSource);
+ try {
+ final List<SupervisorStatus> supervisors = new ArrayList<>();
+
cluster.callApi().onLeaderOverlord(OverlordClient::supervisorStatuses).forEachRemaining(supervisors::add);
+ for (final SupervisorStatus supervisor : supervisors) {
+ if (dataSource.equals(supervisor.getId())) {
+ cluster.callApi().onLeaderOverlord(o ->
o.terminateSupervisor(supervisor.getId()));
+ }
+ }
+
+ cluster.callApi()
+ .waitForResult(this::cancelTasksForCurrentTest, Set::isEmpty)
+ .withTimeoutMillis(60_000)
+ .go();
+ }
+ finally {
+ markSegmentsAsUnused(dataSource);
+ }
+ }
+
+ private Set<String> cancelTasksForCurrentTest()
+ {
+ final Set<String> taskIds = new HashSet<>();
+ for (final String state : List.of("running", "pending", "waiting")) {
+ for (final TaskStatusPlus task : cluster.callApi().getTasks(dataSource,
state)) {
+ taskIds.add(task.getId());
+ }
+ }
+
+ // A task can be complete in storage while still occupying a worker slot.
+ // Docker subclasses may use an external leader, so its runner is not
available here.
+ final Optional<TaskRunner> taskRunner =
overlord.bindings().getInstance(TaskMaster.class).getTaskRunner();
+ if (taskRunner.isPresent()) {
+ final List<TaskRunnerWorkItem> runnerTasks = new
ArrayList<>(taskRunner.get().getPendingTasks());
+ runnerTasks.addAll(taskRunner.get().getRunningTasks());
+ for (final TaskRunnerWorkItem task : runnerTasks) {
+ if (dataSource.equals(task.getDataSource())) {
+ taskIds.add(task.getTaskId());
+ }
+ }
+ }
+
+ for (final String taskId : taskIds) {
+ cluster.callApi().onLeaderOverlord(o -> o.cancelTask(taskId));
+ }
+ return taskIds;
}
protected int markSegmentsAsUnused(String dataSource)
@@ -363,7 +413,7 @@ public class IngestionSmokeTest extends
EmbeddedClusterTestBase
final String taskId = IdUtils.getRandomId();
final long runDurationMillis = 100_000L;
cluster.callApi().onLeaderOverlord(
- o -> o.runTask(taskId, new NoopTask(taskId, null, null,
runDurationMillis, 0L, null))
+ o -> o.runTask(taskId, new NoopTask(taskId, null, dataSource,
runDurationMillis, 0L, null))
);
eventCollector.latchableEmitter().waitForEvent(
diff --git
a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/KafkaBoundedSupervisorTest.java
b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/KafkaBoundedSupervisorTest.java
index 79214ab88c1..7f83375bc0e 100644
---
a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/KafkaBoundedSupervisorTest.java
+++
b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/KafkaBoundedSupervisorTest.java
@@ -29,11 +29,18 @@ import
org.apache.druid.indexing.seekablestream.supervisor.BoundedStreamConfig;
import org.apache.druid.query.DruidMetrics;
import org.apache.druid.testing.embedded.EmbeddedDruidCluster;
import org.apache.druid.testing.embedded.StreamIngestResource;
+import org.apache.druid.testing.embedded.tools.EventSerializer;
+import org.apache.druid.testing.embedded.tools.JsonEventSerializer;
+import org.apache.druid.testing.embedded.tools.StreamGenerator;
+import
org.apache.druid.testing.embedded.tools.WikipediaStreamEventStreamGenerator;
+import org.apache.kafka.clients.producer.ProducerRecord;
import org.joda.time.Period;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.util.ArrayList;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
/**
@@ -203,12 +210,26 @@ public class KafkaBoundedSupervisorTest extends
StreamIndexTestBase
.build(dataSource, topic);
}
+ private void publishRecordsToBothPartitions(String topic)
+ {
+ final EventSerializer serializer = new
JsonEventSerializer(overlord.bindings().jsonMapper());
+ final StreamGenerator generator = new
WikipediaStreamEventStreamGenerator(serializer, 100, 100);
+ final List<byte[]> records = generator.generateEvents(10);
+ final List<ProducerRecord<byte[], byte[]>> producerRecords = new
ArrayList<>();
+ // Fixed per-partition end offsets require data in both partitions;
Kafka's default partitioner need not balance it.
+ for (int i = 0; i < records.size(); i++) {
+ producerRecords.add(new ProducerRecord<>(topic, i % 2, null,
records.get(i)));
+ }
+ kafkaServer.produceRecordsWithoutTransaction(producerRecords);
+ Assertions.assertEquals(Map.of("0", 500L, "1", 500L),
kafkaServer.getPartitionOffsets(topic));
+ }
+
@Test
public void test_boundedSupervisor_withMismatchedMetadata_is_unhealthy()
{
final String topic = IdUtils.getRandomId();
kafkaServer.createTopicWithPartitions(topic, 2);
- publish1kRecords(topic, false);
+ publishRecordsToBothPartitions(topic);
// Get the current end offsets for all partitions
Map<String, Long> currentOffsets = kafkaServer.getPartitionOffsets(topic);
@@ -282,7 +303,7 @@ public class KafkaBoundedSupervisorTest extends
StreamIndexTestBase
{
final String topic = IdUtils.getRandomId();
kafkaServer.createTopicWithPartitions(topic, 2);
- publish1kRecords(topic, false);
+ publishRecordsToBothPartitions(topic);
// Run 1: ingest up to offset 100 on each partition and complete.
Map<String, Long> startOffsets1 = new HashMap<>();
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskQueueScaleTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskQueueScaleTest.java
index 04670ba28a3..e0c21df78b3 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskQueueScaleTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskQueueScaleTest.java
@@ -179,12 +179,8 @@ public class TaskQueueScaleTest
taskQueue.add(testTask);
}
- // in theory we can get a race here, since we fetch the counts at separate
times
+ // The running, pending, and waiting counters are separate snapshots and
cannot be summed while tasks transition.
Assertions.assertEquals(numTasks, taskQueue.getTasks().size(), "all tasks
should be known");
- long runningTasks =
taskQueue.getRunningTaskCount().values().stream().mapToLong(Long::longValue).sum();
- long pendingTasks =
taskQueue.getPendingTaskCount().values().stream().mapToLong(Long::longValue).sum();
- long waitingTasks =
taskQueue.getWaitingTaskCount().values().stream().mapToLong(Long::longValue).sum();
- Assertions.assertEquals(numTasks, (runningTasks + pendingTasks +
waitingTasks), "all tasks should be known");
// Wait for all tasks to finish.
final TaskLookup.CompleteTaskLookup completeTaskLookup =
@@ -194,12 +190,18 @@ public class TaskQueueScaleTest
Thread.sleep(100);
}
- Thread.sleep(100);
+ // Completion is persisted before cleanup finishes. The test timeout
bounds this wait.
+ while (!taskStorage.getActiveTasks().isEmpty()
+ || taskQueue.getRunningTaskCount().values().stream().anyMatch(count
-> count != 0)
+ || taskQueue.getPendingTaskCount().values().stream().anyMatch(count
-> count != 0)
+ || taskQueue.getWaitingTaskCount().values().stream().anyMatch(count
-> count != 0)) {
+ Thread.sleep(100);
+ }
Assertions.assertEquals(0, taskStorage.getActiveTasks().size(), "no tasks
should be active");
- runningTasks =
taskQueue.getRunningTaskCount().values().stream().mapToLong(Long::longValue).sum();
- pendingTasks =
taskQueue.getPendingTaskCount().values().stream().mapToLong(Long::longValue).sum();
- waitingTasks =
taskQueue.getWaitingTaskCount().values().stream().mapToLong(Long::longValue).sum();
+ final long runningTasks =
taskQueue.getRunningTaskCount().values().stream().mapToLong(Long::longValue).sum();
+ final long pendingTasks =
taskQueue.getPendingTaskCount().values().stream().mapToLong(Long::longValue).sum();
+ final long waitingTasks =
taskQueue.getWaitingTaskCount().values().stream().mapToLong(Long::longValue).sum();
Assertions.assertEquals(0, runningTasks, "no tasks should be running");
Assertions.assertEquals(0, pendingTasks, "no tasks should be pending");
Assertions.assertEquals(0, waitingTasks, "no tasks should be waiting");
diff --git
a/server/src/main/java/org/apache/druid/segment/metadata/AbstractSegmentMetadataCache.java
b/server/src/main/java/org/apache/druid/segment/metadata/AbstractSegmentMetadataCache.java
index d8677fba952..34d1e755e2f 100644
---
a/server/src/main/java/org/apache/druid/segment/metadata/AbstractSegmentMetadataCache.java
+++
b/server/src/main/java/org/apache/druid/segment/metadata/AbstractSegmentMetadataCache.java
@@ -591,7 +591,7 @@ public abstract class AbstractSegmentMetadataCache<T
extends DataSourceInformati
segmentsNeedingRefresh.remove(segment.getId());
unmarkSegmentAsMutable(segment.getId());
- segmentMetadataInfo.compute(
+ final ConcurrentSkipListMap<SegmentId, AvailableSegmentMetadata>
remainingSegments = segmentMetadataInfo.compute(
segment.getDataSource(),
(dataSource, segmentsMap) -> {
if (segmentsMap == null) {
@@ -605,7 +605,11 @@ public abstract class AbstractSegmentMetadataCache<T
extends DataSourceInformati
}
removeSegmentAction(segment.getId());
if (segmentsMap.isEmpty()) {
- tables.remove(segment.getDataSource());
+ // Emit the removal action only if this call actually removed
the table, so that a concurrent
+ // refresh which also finds the datasource gone cannot report
the same removal twice.
+ if (tables.remove(segment.getDataSource()) != null) {
+ removeDataSourceAction(segment.getDataSource());
+ }
log.info("dataSource [%s] no longer exists, all metadata
removed.", segment.getDataSource());
return null;
} else {
@@ -615,6 +619,9 @@ public abstract class AbstractSegmentMetadataCache<T
extends DataSourceInformati
}
}
);
+ if (remainingSegments == null) {
+ dataSourcesNeedingRebuild.remove(segment.getDataSource());
+ }
lock.notifyAll();
}
@@ -625,6 +632,16 @@ public abstract class AbstractSegmentMetadataCache<T
extends DataSourceInformati
*/
protected abstract void removeSegmentAction(SegmentId segmentId);
+ /**
+ * Called under the cache lock after the last segment of a datasource has
been removed and its table was actually
+ * removed from {@link #tables} by that removal. It is not called when no
table existed for the datasource, so a
+ * single datasource removal triggers this action at most once even if a
refresh observes the removal concurrently.
+ */
+ protected void removeDataSourceAction(String dataSource)
+ {
+ // No additional action by default.
+ }
+
@VisibleForTesting
public void removeServerSegment(final DruidServerMetadata server, final
DataSegment segment)
{
diff --git
a/sql/src/main/java/org/apache/druid/sql/calcite/schema/BrokerSegmentMetadataCache.java
b/sql/src/main/java/org/apache/druid/sql/calcite/schema/BrokerSegmentMetadataCache.java
index 26a6c63660d..0baa5294f56 100644
---
a/sql/src/main/java/org/apache/druid/sql/calcite/schema/BrokerSegmentMetadataCache.java
+++
b/sql/src/main/java/org/apache/druid/sql/calcite/schema/BrokerSegmentMetadataCache.java
@@ -251,11 +251,11 @@ public class BrokerSegmentMetadataCache extends
AbstractSegmentMetadataCache<Phy
final RowSignature rowSignature =
buildDataSourceRowSignature(dataSource);
if (rowSignature == null) {
log.info("datasource [%s] no longer exists, all metadata removed.",
dataSource);
- tables.remove(dataSource);
- emitMetric(
- Metric.DATASOURCE_REMOVED,
- 1,
- ServiceMetricEvent.builder().setDimension(DruidMetrics.DATASOURCE,
dataSource));
+ // The last-segment callback may already have removed the table and
emitted the metric while this refresh
+ // was in flight. Only emit if this refresh is the one that actually
removed the table.
+ if (tables.remove(dataSource) != null) {
+ emitDataSourceRemoved(dataSource);
+ }
continue;
}
@@ -264,12 +264,11 @@ public class BrokerSegmentMetadataCache extends
AbstractSegmentMetadataCache<Phy
// and a new datasource is added
log.info("datasource [%s] schema has not been initialized yet, "
+ "check coordinator logs if this message is persistent.",
dataSource);
- // this is a harmless call
- tables.remove(dataSource);
- emitMetric(
- Metric.DATASOURCE_REMOVED,
- 1,
- ServiceMetricEvent.builder().setDimension(DruidMetrics.DATASOURCE,
dataSource));
+ // Usually there is no table to remove here. If there was one, a
concurrent last-segment callback may have
+ // removed it and emitted the metric already, so only emit if this
refresh actually removed the table.
+ if (tables.remove(dataSource) != null) {
+ emitDataSourceRemoved(dataSource);
+ }
continue;
}
@@ -307,6 +306,22 @@ public class BrokerSegmentMetadataCache extends
AbstractSegmentMetadataCache<Phy
// noop, no additional action needed when segment is removed.
}
+ @Override
+ protected void removeDataSourceAction(String dataSource)
+ {
+ // The last-segment callback can remove the table without another schema
refresh.
+ emitDataSourceRemoved(dataSource);
+ }
+
+ private void emitDataSourceRemoved(String dataSource)
+ {
+ emitMetric(
+ Metric.DATASOURCE_REMOVED,
+ 1,
+ ServiceMetricEvent.builder().setDimension(DruidMetrics.DATASOURCE,
dataSource)
+ );
+ }
+
private Set<String> queryDataSources()
{
Set<String> dataSources = new HashSet<>();
diff --git
a/sql/src/test/java/org/apache/druid/sql/calcite/schema/BrokerSegmentMetadataCacheTest.java
b/sql/src/test/java/org/apache/druid/sql/calcite/schema/BrokerSegmentMetadataCacheTest.java
index 55abaf7b4a3..57e10f07a5b 100644
---
a/sql/src/test/java/org/apache/druid/sql/calcite/schema/BrokerSegmentMetadataCacheTest.java
+++
b/sql/src/test/java/org/apache/druid/sql/calcite/schema/BrokerSegmentMetadataCacheTest.java
@@ -34,6 +34,7 @@ import org.apache.calcite.rel.type.RelDataTypeField;
import org.apache.calcite.sql.type.SqlTypeName;
import org.apache.druid.client.ImmutableDruidServer;
import org.apache.druid.client.InternalQueryConfig;
+import org.apache.druid.client.TimelineServerView;
import org.apache.druid.client.coordinator.CoordinatorClient;
import org.apache.druid.client.coordinator.NoopCoordinatorClient;
import org.apache.druid.data.input.InputRow;
@@ -105,6 +106,8 @@ import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
public class BrokerSegmentMetadataCacheTest extends
BrokerSegmentMetadataCacheTestBase
@@ -720,6 +723,168 @@ public class BrokerSegmentMetadataCacheTest extends
BrokerSegmentMetadataCacheTe
Assertions.assertEquals(5, schema.getSegmentMetadataSnapshot().size());
}
+ @Test
+ public void testLastSegmentRemovalClearsRebuildState() throws IOException
+ {
+ final BrokerSegmentMetadataCache schema = new BrokerSegmentMetadataCache(
+ CalciteTests.createMockQueryLifecycleFactory(walker, conglomerate),
+ Mockito.mock(TimelineServerView.class),
+ SEGMENT_CACHE_CONFIG_DEFAULT,
+ new NoopEscalator(),
+ new InternalQueryConfig(),
+ emitter,
+ new PhysicalDatasourceMetadataFactory(globalTableJoinable,
segmentManager),
+ new NoopCoordinatorClient(),
+ CentralizedDatasourceSchemaConfig.create()
+ );
+ runningSchema = schema;
+ final List<DataSegment> segments = ImmutableList.of(segment1, segment2);
+ segments.forEach(segment ->
schema.addSegment(druidServers.get(0).getMetadata(), segment));
+ schema.refresh(
+ segments.stream().map(DataSegment::getId).collect(Collectors.toSet()),
+ new HashSet<>(Set.of("foo"))
+ );
+ Assertions.assertNotNull(schema.getDatasource("foo"));
+ emitter.flush();
+
+ // Removing a non-last segment marks the datasource for rebuild before the
last segment is removed.
+ schema.removeSegment(segments.get(0));
+ schema.removeSegment(segments.get(1));
+ Assertions.assertNull(schema.getDatasource("foo"));
+ emitter.verifyEmitted(Metric.DATASOURCE_REMOVED,
Map.of(DruidMetrics.DATASOURCE, "foo"), 1);
+
+ // A later refresh must not process stale rebuild state and emit the
removal metric again.
+ schema.refresh(new HashSet<>(), new HashSet<>());
+ emitter.verifyEmitted(Metric.DATASOURCE_REMOVED,
Map.of(DruidMetrics.DATASOURCE, "foo"), 1);
+
+ // A refresh that had already captured the datasource for rebuild before
the last segment was removed
+ // (or that is handed the datasource explicitly) finds no table left to
remove and must not emit again.
+ schema.refresh(new HashSet<>(), new HashSet<>(Set.of("foo")));
+ Assertions.assertNull(schema.getDatasource("foo"));
+ emitter.verifyEmitted(Metric.DATASOURCE_REMOVED,
Map.of(DruidMetrics.DATASOURCE, "foo"), 1);
+ }
+
+ @Test
+ public void testRefreshOfUnknownDatasourceDoesNotEmitRemovalMetric() throws
IOException
+ {
+ final BrokerSegmentMetadataCache schema = new BrokerSegmentMetadataCache(
+ CalciteTests.createMockQueryLifecycleFactory(walker, conglomerate),
+ Mockito.mock(TimelineServerView.class),
+ SEGMENT_CACHE_CONFIG_DEFAULT,
+ new NoopEscalator(),
+ new InternalQueryConfig(),
+ emitter,
+ new PhysicalDatasourceMetadataFactory(globalTableJoinable,
segmentManager),
+ new NoopCoordinatorClient(),
+ CentralizedDatasourceSchemaConfig.create()
+ );
+ runningSchema = schema;
+
+ // No table was ever built for this datasource, so there is nothing to
report as removed.
+ schema.refresh(new HashSet<>(), new HashSet<>(Set.of("never-existed")));
+ Assertions.assertNull(schema.getDatasource("never-existed"));
+ emitter.verifyNotEmitted(Metric.DATASOURCE_REMOVED);
+ }
+
+ @Test
+ public void testRefreshOfUninitializedDatasourceDoesNotEmitRemovalMetric()
throws IOException
+ {
+ final BrokerSegmentMetadataCache schema = new BrokerSegmentMetadataCache(
+ CalciteTests.createMockQueryLifecycleFactory(walker, conglomerate),
+ Mockito.mock(TimelineServerView.class),
+ SEGMENT_CACHE_CONFIG_DEFAULT,
+ new NoopEscalator(),
+ new InternalQueryConfig(),
+ emitter,
+ new PhysicalDatasourceMetadataFactory(globalTableJoinable,
segmentManager),
+ new NoopCoordinatorClient(),
+ CentralizedDatasourceSchemaConfig.create()
+ );
+ runningSchema = schema;
+
+ // The segment is known but has not been refreshed, so the datasource row
signature is empty and no table has
+ // ever been built. Rebuilding it must not report a removal.
+ schema.addSegment(druidServers.get(0).getMetadata(), segment1);
+ schema.refresh(new HashSet<>(), new HashSet<>(Set.of("foo")));
+ Assertions.assertNull(schema.getDatasource("foo"));
+ emitter.verifyNotEmitted(Metric.DATASOURCE_REMOVED);
+ }
+
+ @Test
+ public void
testLastSegmentRemovedDuringRefreshWithUninitializedSegmentEmitsRemovalOnce()
throws Exception
+ {
+ final AtomicBoolean blockSignatureBuild = new AtomicBoolean(false);
+ final CountDownLatch signatureBuilt = new CountDownLatch(1);
+ final CountDownLatch resumeRefresh = new CountDownLatch(1);
+ final BrokerSegmentMetadataCache schema = new BrokerSegmentMetadataCache(
+ CalciteTests.createMockQueryLifecycleFactory(walker, conglomerate),
+ Mockito.mock(TimelineServerView.class),
+ SEGMENT_CACHE_CONFIG_DEFAULT,
+ new NoopEscalator(),
+ new InternalQueryConfig(),
+ emitter,
+ new PhysicalDatasourceMetadataFactory(globalTableJoinable,
segmentManager),
+ new NoopCoordinatorClient(),
+ CentralizedDatasourceSchemaConfig.create()
+ )
+ {
+ @Override
+ public RowSignature buildDataSourceRowSignature(String dataSource)
+ {
+ final RowSignature rowSignature =
super.buildDataSourceRowSignature(dataSource);
+ if (blockSignatureBuild.get()) {
+ // Hold the refresh between computing the (empty) signature and
acting on it.
+ signatureBuilt.countDown();
+ try {
+ Assertions.assertTrue(resumeRefresh.await(WAIT_TIMEOUT_SECS,
TimeUnit.SECONDS));
+ }
+ catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ }
+ return rowSignature;
+ }
+ };
+ runningSchema = schema;
+
+ // Build the table from a refreshed segment.
+ schema.addSegment(druidServers.get(0).getMetadata(), segment1);
+ schema.refresh(new HashSet<>(Set.of(segment1.getId())), new
HashSet<>(Set.of("foo")));
+ Assertions.assertNotNull(schema.getDatasource("foo"));
+ emitter.flush();
+
+ // Leave only an unrefreshed segment behind, so the next rebuild sees an
empty (non-null) row signature.
+ schema.addSegment(druidServers.get(0).getMetadata(), segment2);
+ schema.removeSegment(segment1);
+
+ blockSignatureBuild.set(true);
+ final AtomicReference<Throwable> refreshError = new AtomicReference<>();
+ final Thread refreshThread = new Thread(() -> {
+ try {
+ schema.refresh(new HashSet<>(), new HashSet<>(Set.of("foo")));
+ }
+ catch (Throwable t) {
+ refreshError.set(t);
+ }
+ });
+ refreshThread.start();
+ Assertions.assertTrue(signatureBuilt.await(WAIT_TIMEOUT_SECS,
TimeUnit.SECONDS));
+
+ // The last segment disappears while the refresh is in flight: the
callback removes the table and emits once.
+ schema.removeSegment(segment2);
+ Assertions.assertNull(schema.getDatasource("foo"));
+ emitter.verifyEmitted(Metric.DATASOURCE_REMOVED,
Map.of(DruidMetrics.DATASOURCE, "foo"), 1);
+
+ // The refresh resumes, finds the table already gone and must not emit a
second removal.
+ resumeRefresh.countDown();
+ refreshThread.join(TimeUnit.SECONDS.toMillis(WAIT_TIMEOUT_SECS));
+ Assertions.assertFalse(refreshThread.isAlive(), "refresh did not finish");
+ Assertions.assertNull(refreshError.get());
+ Assertions.assertNull(schema.getDatasource("foo"));
+ emitter.verifyEmitted(Metric.DATASOURCE_REMOVED,
Map.of(DruidMetrics.DATASOURCE, "foo"), 1);
+ }
+
@Test
public void testAllDatasourcesRebuiltOnDatasourceRemoval() throws
IOException, InterruptedException
{
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
