This is an automated email from the ASF dual-hosted git repository.
gianm 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 3e8da56c809 fix: dedicated ephemeral storage loading pool for tasks
(#19658)
3e8da56c809 is described below
commit 3e8da56c80992f472dee2690010d49ebbef6ced8
Author: Clint Wylie <[email protected]>
AuthorDate: Mon Jul 20 11:24:12 2026 -0700
fix: dedicated ephemeral storage loading pool for tasks (#19658)
---
.../druid/guice/ClusterTestingModuleTest.java | 2 +-
.../common/SegmentCacheManagerFactory.java | 43 ++++++--
.../common/SegmentCacheManagerFactoryTest.java | 109 +++++++++++++++++++++
.../task/ClientCompactionTaskQuerySerdeTest.java | 4 +-
.../common/task/CompactionTaskRunBase.java | 2 +-
.../indexing/common/task/CompactionTaskTest.java | 4 +-
.../indexing/common/task/IngestionTestBase.java | 2 +-
.../druid/indexing/common/task/TaskBuilder.java | 2 +-
.../AbstractMultiPhaseParallelIndexingTest.java | 2 +-
.../AbstractParallelIndexSupervisorTaskTest.java | 2 +-
.../compact/OverlordCompactionSchedulerTest.java | 4 +-
.../overlord/SingleTaskBackgroundRunnerTest.java | 2 +-
.../druid/indexing/overlord/TaskLifecycleTest.java | 2 +-
.../indexing/overlord/TestTaskToolboxFactory.java | 2 +-
.../SeekableStreamIndexTaskTestBase.java | 2 +-
.../indexing/worker/WorkerTaskManagerTest.java | 2 +-
.../druid/msq/test/CalciteMSQTestsHelper.java | 2 +-
.../org/apache/druid/msq/test/MSQTestBase.java | 2 +-
.../guice/annotations/EphemeralStorageLoading.java | 37 +++++++
.../org/apache/druid/guice/StorageNodeModule.java | 15 +++
.../druid/segment/loading/SegmentLoaderConfig.java | 41 ++++++++
.../segment/loading/StorageLoadingThreadPool.java | 3 +-
.../apache/druid/guice/StorageNodeModuleTest.java | 22 +++++
.../segment/loading/SegmentLoaderConfigTest.java | 18 ++++
.../loading/StorageLoadingThreadPoolTest.java | 62 ++++++++++++
.../java/org/apache/druid/cli/CliPeonTest.java | 2 +-
26 files changed, 361 insertions(+), 29 deletions(-)
diff --git
a/extensions-core/testing-tools/src/test/java/org/apache/druid/guice/ClusterTestingModuleTest.java
b/extensions-core/testing-tools/src/test/java/org/apache/druid/guice/ClusterTestingModuleTest.java
index 63decde23a5..9e98dfcda87 100644
---
a/extensions-core/testing-tools/src/test/java/org/apache/druid/guice/ClusterTestingModuleTest.java
+++
b/extensions-core/testing-tools/src/test/java/org/apache/druid/guice/ClusterTestingModuleTest.java
@@ -191,7 +191,7 @@ public class ClusterTestingModuleTest
null,
indexIO,
new NoopCoordinatorClient(),
- new SegmentCacheManagerFactory(indexIO, MAPPER),
+ SegmentCacheManagerFactory.createWithOwnedPool(indexIO, MAPPER),
new TaskConfigBuilder().build()
);
final ParallelIndexIOConfig ioConfig = new ParallelIndexIOConfig(
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/common/SegmentCacheManagerFactory.java
b/indexing-service/src/main/java/org/apache/druid/indexing/common/SegmentCacheManagerFactory.java
index cf5dfc7d0e9..402e4be10c5 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/common/SegmentCacheManagerFactory.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/common/SegmentCacheManagerFactory.java
@@ -21,8 +21,12 @@ package org.apache.druid.indexing.common;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.inject.Inject;
+import com.google.inject.Provider;
+import com.google.inject.util.Providers;
+import org.apache.druid.guice.annotations.EphemeralStorageLoading;
import org.apache.druid.guice.annotations.Json;
import org.apache.druid.segment.IndexIO;
+import org.apache.druid.segment.loading.AcquireMode;
import
org.apache.druid.segment.loading.LeastBytesUsedStorageLocationSelectorStrategy;
import org.apache.druid.segment.loading.SegmentCacheManager;
import org.apache.druid.segment.loading.SegmentLoaderConfig;
@@ -43,26 +47,52 @@ public class SegmentCacheManagerFactory
{
private final IndexIO indexIO;
private final ObjectMapper jsonMapper;
+ /**
+ * Resolved lazily (only on the {@code virtualStorage=true} path of {@link
#manufacturate}) so that injecting this
+ * factory into a component that only builds non-virtual caches (e.g. {@code
DruidInputSource}) does not force the
+ * process-wide {@link EphemeralStorageLoading} pool to be created on
processes that never do on-demand loading
+ * (Overlord, MiddleManager).
+ */
+ private final Provider<StorageLoadingThreadPool> loadingThreadPoolProvider;
@Inject
public SegmentCacheManagerFactory(
IndexIO indexIO,
- @Json ObjectMapper mapper
+ @Json ObjectMapper mapper,
+ @EphemeralStorageLoading Provider<StorageLoadingThreadPool>
loadingThreadPoolProvider
)
{
this.indexIO = indexIO;
this.jsonMapper = mapper;
+ this.loadingThreadPoolProvider = loadingThreadPoolProvider;
+ }
+
+ /**
+ * Creates a factory whose {@code virtualStorage=true} managers get their
own private loading pool of default size,
+ * instead of the process-wide, lifecycle-managed {@link
EphemeralStorageLoading} pool that Guice injects. Intended
+ * for tests and manual construction where that shared pool is not available
via injection; the created pool is not
+ * lifecycle-managed, which is fine for short-lived JVMs.
+ */
+ public static SegmentCacheManagerFactory createWithOwnedPool(IndexIO
indexIO, ObjectMapper mapper)
+ {
+ return new SegmentCacheManagerFactory(
+ indexIO,
+ mapper,
+ Providers.of(StorageLoadingThreadPool.createFromConfig(new
SegmentLoaderConfig().setVirtualStorage(true)))
+ );
}
/**
- * Creates a new {@link SegmentCacheManager} backed by a new storage
location in {@code storageDir}, and a new
- * loading thread pool of default size.
+ * Creates a new {@link SegmentCacheManager} backed by a new storage
location in {@code storageDir}. When
+ * {@code virtualStorage} is true, the returned manager uses this factory's
process-wide
+ * {@link EphemeralStorageLoading} loading pool (shared across all per-task
caches and stopped by the lifecycle)
+ * rather than creating its own.
*
* @param storageDir storage location
* @param maxSize size limit, or null for no limit
* @param virtualStorage whether to configure the cache manager in ephemeral
virtual storage mode. In this mode,
- * loading is triggered by {@link
SegmentCacheManager#acquireSegment(DataSegment)}, and
- * segment files are deleted as soon as all holds are
closed.
+ * loading is triggered by {@link
SegmentCacheManager#acquireSegment(DataSegment, AcquireMode)},
+ * and segment files are deleted as soon as all holds
are closed.
*/
public SegmentCacheManager manufacturate(File storageDir, Long maxSize,
boolean virtualStorage)
{
@@ -77,11 +107,10 @@ public class SegmentCacheManagerFactory
.setVirtualStorage(virtualStorage)
.setVirtualStorageIsEphemeral(virtualStorage);
final List<StorageLocation> storageLocations =
loaderConfig.toStorageLocations();
- final StorageLoadingThreadPool loadingThreadPool =
StorageLoadingThreadPool.createFromConfig(loaderConfig);
return new SegmentLocalCacheManager(
storageLocations,
loaderConfig,
- loadingThreadPool,
+ virtualStorage ? loadingThreadPoolProvider.get() :
StorageLoadingThreadPool.none(),
new LeastBytesUsedStorageLocationSelectorStrategy(storageLocations),
indexIO,
jsonMapper
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/common/SegmentCacheManagerFactoryTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/common/SegmentCacheManagerFactoryTest.java
new file mode 100644
index 00000000000..87622425794
--- /dev/null
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/common/SegmentCacheManagerFactoryTest.java
@@ -0,0 +1,109 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.indexing.common;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.inject.Provider;
+import com.google.inject.util.Providers;
+import org.apache.druid.java.util.emitter.EmittingLogger;
+import org.apache.druid.segment.TestHelper;
+import org.apache.druid.segment.TestIndex;
+import org.apache.druid.segment.loading.SegmentCacheManager;
+import org.apache.druid.segment.loading.SegmentLoaderConfig;
+import org.apache.druid.segment.loading.StorageLoadingThreadPool;
+import org.apache.druid.server.metrics.NoopServiceEmitter;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+
+class SegmentCacheManagerFactoryTest
+{
+ @TempDir
+ File tempDir;
+
+ private ObjectMapper jsonMapper;
+
+ @BeforeAll
+ static void setUpClass()
+ {
+ EmittingLogger.registerEmitter(new NoopServiceEmitter());
+ }
+
+ @BeforeEach
+ void setUp()
+ {
+ jsonMapper = TestHelper.makeJsonMapper();
+ }
+
+ @Test
+ void testVirtualStorageManagersShareTheInjectedPool()
+ {
+ final StorageLoadingThreadPool shared =
+ StorageLoadingThreadPool.createFromConfig(new
SegmentLoaderConfig().withVirtualStorage(true));
+ try {
+ final SegmentCacheManagerFactory factory =
+ new SegmentCacheManagerFactory(TestIndex.INDEX_IO, jsonMapper,
Providers.of(shared));
+ final SegmentCacheManager m1 = factory.manufacturate(new File(tempDir,
"a"), null, true);
+ final SegmentCacheManager m2 = factory.manufacturate(new File(tempDir,
"b"), null, true);
+
+ Assertions.assertSame(shared, m1.getLoadingThreadPool());
+ Assertions.assertSame(shared, m2.getLoadingThreadPool());
+ }
+ finally {
+ shared.stop();
+ }
+ }
+
+ @Test
+ void testNonVirtualStorageDoesNotResolveTheLoadingPool()
+ {
+ // Injecting the factory into a virtualStorage=false-only consumer (e.g.
DruidInputSource) must not force the
+ // ephemeral loading pool to be created. The provider is only resolved on
the virtualStorage=true path.
+ final Provider<StorageLoadingThreadPool> throwingProvider = () -> {
+ throw new AssertionError("ephemeral loading pool must not be resolved
for virtualStorage=false");
+ };
+ final SegmentCacheManagerFactory factory =
+ new SegmentCacheManagerFactory(TestIndex.INDEX_IO, jsonMapper,
throwingProvider);
+
+ final SegmentCacheManager m = factory.manufacturate(new File(tempDir,
"c"), null, false);
+ Assertions.assertFalse(m.getLoadingThreadPool().isAvailable());
+ }
+
+ @Test
+ void testCreateWithOwnedPoolBuildsOneAvailablePoolPerFactory()
+ {
+ final SegmentCacheManagerFactory factory =
+ SegmentCacheManagerFactory.createWithOwnedPool(TestIndex.INDEX_IO,
jsonMapper);
+ final SegmentCacheManager m1 = factory.manufacturate(new File(tempDir,
"d"), null, true);
+ final SegmentCacheManager m2 = factory.manufacturate(new File(tempDir,
"e"), null, true);
+ try {
+ Assertions.assertTrue(m1.getLoadingThreadPool().isAvailable());
+ // createWithOwnedPool builds one pool and shares it across the
factory's manufacturate calls.
+ Assertions.assertSame(m1.getLoadingThreadPool(),
m2.getLoadingThreadPool());
+ }
+ finally {
+ m1.getLoadingThreadPool().stop();
+ }
+ }
+}
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/ClientCompactionTaskQuerySerdeTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/ClientCompactionTaskQuerySerdeTest.java
index f33548e36b3..4427ddb0d93 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/ClientCompactionTaskQuerySerdeTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/ClientCompactionTaskQuerySerdeTest.java
@@ -215,7 +215,7 @@ public class ClientCompactionTaskQuerySerdeTest
binder.bind(ChatHandlerProvider.class).toInstance(new
ChatHandlerProvider());
binder.bind(RowIngestionMetersFactory.class).toInstance(ROW_INGESTION_METERS_FACTORY);
binder.bind(CoordinatorClient.class).toInstance(COORDINATOR_CLIENT);
- binder.bind(SegmentCacheManagerFactory.class).toInstance(new
SegmentCacheManagerFactory(TestIndex.INDEX_IO, objectMapper));
+
binder.bind(SegmentCacheManagerFactory.class).toInstance(SegmentCacheManagerFactory.createWithOwnedPool(TestIndex.INDEX_IO,
objectMapper));
binder.bind(AppenderatorsManager.class).toInstance(APPENDERATORS_MANAGER);
binder.bind(OverlordClient.class).toInstance(new
NoopOverlordClient());
}
@@ -379,7 +379,7 @@ public class ClientCompactionTaskQuerySerdeTest
{
CompactionTask.Builder compactionTaskBuilder = new CompactionTask.Builder(
"datasource",
- new SegmentCacheManagerFactory(TestIndex.INDEX_IO, MAPPER)
+ SegmentCacheManagerFactory.createWithOwnedPool(TestIndex.INDEX_IO,
MAPPER)
)
.inputSpec(new CompactionIntervalSpec(Intervals.of("2019/2020"),
"testSha256OfSortedSegmentIds"), true)
.tuningConfig(
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskRunBase.java
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskRunBase.java
index 7d20385c465..267cc906881 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskRunBase.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskRunBase.java
@@ -233,7 +233,7 @@ public abstract class CompactionTaskRunBase
temporaryFolder.create();
reportsFile = temporaryFolder.newFile();
testUtils = new TestUtils();
- segmentCacheManagerFactory = new
SegmentCacheManagerFactory(TestIndex.INDEX_IO, testUtils.getTestObjectMapper());
+ segmentCacheManagerFactory =
SegmentCacheManagerFactory.createWithOwnedPool(TestIndex.INDEX_IO,
testUtils.getTestObjectMapper());
objectMapper = testUtils.getTestObjectMapper();
objectMapper.registerSubtypes(new NamedType(LocalLoadSpec.class, "local"));
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskTest.java
index 5c8a49ed482..8a8d403e160 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskTest.java
@@ -309,7 +309,7 @@ public class CompactionTaskTest
binder.bind(RowIngestionMetersFactory.class).toInstance(TEST_UTILS.getRowIngestionMetersFactory());
binder.bind(CoordinatorClient.class).toInstance(COORDINATOR_CLIENT);
binder.bind(SegmentCacheManagerFactory.class)
- .toInstance(new
SegmentCacheManagerFactory(TestIndex.INDEX_IO, objectMapper));
+
.toInstance(SegmentCacheManagerFactory.createWithOwnedPool(TestIndex.INDEX_IO,
objectMapper));
binder.bind(AppenderatorsManager.class).toInstance(new
TestAppenderatorsManager());
binder.bind(ExprMacroTable.class).toInstance(TestExprMacroTable.INSTANCE);
}
@@ -377,7 +377,7 @@ public class CompactionTaskTest
testIndexIO,
SEGMENT_MAP
);
- segmentCacheManagerFactory = new
SegmentCacheManagerFactory(TestIndex.INDEX_IO, OBJECT_MAPPER);
+ segmentCacheManagerFactory =
SegmentCacheManagerFactory.createWithOwnedPool(TestIndex.INDEX_IO,
OBJECT_MAPPER);
}
@Test
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/IngestionTestBase.java
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/IngestionTestBase.java
index 7ca28f7f3b7..7673350e802 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/IngestionTestBase.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/IngestionTestBase.java
@@ -187,7 +187,7 @@ public abstract class IngestionTestBase extends
InitializedNullHandlingTest
);
lockbox = new GlobalTaskLockbox(taskStorage, storageCoordinator);
lockbox.syncFromStorage();
- segmentCacheManagerFactory = new
SegmentCacheManagerFactory(TestIndex.INDEX_IO, getObjectMapper());
+ segmentCacheManagerFactory =
SegmentCacheManagerFactory.createWithOwnedPool(TestIndex.INDEX_IO,
getObjectMapper());
reportsFile = temporaryFolder.newFile();
dataSegmentKiller = new TestDataSegmentKiller();
taskActionToolbox = createTaskActionToolbox();
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/TaskBuilder.java
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/TaskBuilder.java
index e5040904a90..c3bcdcab0b0 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/TaskBuilder.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/TaskBuilder.java
@@ -144,7 +144,7 @@ public abstract class TaskBuilder<
null,
TestHelper.getTestIndexIO(),
new NoopCoordinatorClient(),
- new SegmentCacheManagerFactory(TestHelper.getTestIndexIO(),
TestHelper.JSON_MAPPER),
+
SegmentCacheManagerFactory.createWithOwnedPool(TestHelper.getTestIndexIO(),
TestHelper.JSON_MAPPER),
new TaskConfigBuilder().build()
)
);
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/AbstractMultiPhaseParallelIndexingTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/AbstractMultiPhaseParallelIndexingTest.java
index abd74439d25..ef0a730e932 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/AbstractMultiPhaseParallelIndexingTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/AbstractMultiPhaseParallelIndexingTest.java
@@ -257,7 +257,7 @@ abstract class AbstractMultiPhaseParallelIndexingTest
extends AbstractParallelIn
private Segment loadSegment(DataSegment dataSegment, File tempSegmentDir)
{
- final SegmentCacheManager cacheManager = new
SegmentCacheManagerFactory(TestIndex.INDEX_IO, getObjectMapper())
+ final SegmentCacheManager cacheManager =
SegmentCacheManagerFactory.createWithOwnedPool(TestIndex.INDEX_IO,
getObjectMapper())
.manufacturate(tempSegmentDir, null, false);
try {
cacheManager.load(dataSegment);
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/AbstractParallelIndexSupervisorTaskTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/AbstractParallelIndexSupervisorTaskTest.java
index e147d032ad5..123003a7ad3 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/AbstractParallelIndexSupervisorTaskTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/AbstractParallelIndexSupervisorTaskTest.java
@@ -636,7 +636,7 @@ public class AbstractParallelIndexSupervisorTaskTest
extends IngestionTestBase
.addValue(AppenderatorsManager.class,
TestUtils.APPENDERATORS_MANAGER)
.addValue(LocalDataSegmentPuller.class, new
LocalDataSegmentPuller())
.addValue(CoordinatorClient.class, coordinatorClient)
- .addValue(SegmentCacheManagerFactory.class, new
SegmentCacheManagerFactory(TestIndex.INDEX_IO, objectMapper))
+ .addValue(SegmentCacheManagerFactory.class,
SegmentCacheManagerFactory.createWithOwnedPool(TestIndex.INDEX_IO,
objectMapper))
.addValue(RetryPolicyFactory.class, new RetryPolicyFactory(new
RetryPolicyConfig()))
.addValue(TaskConfig.class, taskConfig)
);
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/compact/OverlordCompactionSchedulerTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/compact/OverlordCompactionSchedulerTest.java
index 3b9597cd79f..4073221223f 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/compact/OverlordCompactionSchedulerTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/compact/OverlordCompactionSchedulerTest.java
@@ -115,7 +115,7 @@ public class OverlordCompactionSchedulerTest
.Std()
.addValue(
SegmentCacheManagerFactory.class,
- new SegmentCacheManagerFactory(TestIndex.INDEX_IO,
OBJECT_MAPPER)
+
SegmentCacheManagerFactory.createWithOwnedPool(TestIndex.INDEX_IO,
OBJECT_MAPPER)
)
);
}
@@ -230,7 +230,7 @@ public class OverlordCompactionSchedulerTest
TestIndex.INDEX_IO,
Mockito.mock(TaskConfig.class),
new NoopCoordinatorClient(),
- new SegmentCacheManagerFactory(TestIndex.INDEX_IO, OBJECT_MAPPER)
+ SegmentCacheManagerFactory.createWithOwnedPool(TestIndex.INDEX_IO,
OBJECT_MAPPER)
),
(nameFormat, numThreads) -> new
WrappingScheduledExecutorService("test", executor, false),
brokerClient,
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/SingleTaskBackgroundRunnerTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/SingleTaskBackgroundRunnerTest.java
index 37b2c5323b3..b628f8a253c 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/SingleTaskBackgroundRunnerTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/SingleTaskBackgroundRunnerTest.java
@@ -117,7 +117,7 @@ public class SingleTaskBackgroundRunnerTest
null,
NoopJoinableFactory.INSTANCE,
null,
- new SegmentCacheManagerFactory(TestIndex.INDEX_IO,
utils.getTestObjectMapper()),
+ SegmentCacheManagerFactory.createWithOwnedPool(TestIndex.INDEX_IO,
utils.getTestObjectMapper()),
utils.getTestObjectMapper(),
utils.getTestIndexIO(),
null,
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskLifecycleTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskLifecycleTest.java
index ef723b94ee5..5da3d2ed754 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskLifecycleTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskLifecycleTest.java
@@ -581,7 +581,7 @@ public class TaskLifecycleTest extends
InitializedNullHandlingTest
DirectQueryProcessingPool.INSTANCE, // query executor service
NoopJoinableFactory.INSTANCE,
() -> monitorScheduler, // monitor scheduler
- new SegmentCacheManagerFactory(TestIndex.INDEX_IO, new
DefaultObjectMapper()),
+ SegmentCacheManagerFactory.createWithOwnedPool(TestIndex.INDEX_IO, new
DefaultObjectMapper()),
MAPPER,
INDEX_IO,
MapCache.create(0),
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TestTaskToolboxFactory.java
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TestTaskToolboxFactory.java
index b566252f56d..de263efd871 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TestTaskToolboxFactory.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TestTaskToolboxFactory.java
@@ -148,7 +148,7 @@ public class TestTaskToolboxFactory extends
TaskToolboxFactory
private Provider<MonitorScheduler> monitorSchedulerProvider;
private ObjectMapper jsonMapper = TestHelper.JSON_MAPPER;
private IndexIO indexIO = TestHelper.getTestIndexIO();
- private SegmentCacheManagerFactory segmentCacheManagerFactory = new
SegmentCacheManagerFactory(TestIndex.INDEX_IO, jsonMapper);
+ private SegmentCacheManagerFactory segmentCacheManagerFactory =
SegmentCacheManagerFactory.createWithOwnedPool(TestIndex.INDEX_IO, jsonMapper);
private Cache cache;
private CacheConfig cacheConfig;
private CachePopulatorStats cachePopulatorStats;
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskTestBase.java
b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskTestBase.java
index ce48c0efda2..1821fc5b38c 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskTestBase.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskTestBase.java
@@ -681,7 +681,7 @@ public abstract class SeekableStreamIndexTaskTestBase
extends EasyMockSupport
DirectQueryProcessingPool.INSTANCE,
NoopJoinableFactory.INSTANCE,
() -> EasyMock.createMock(MonitorScheduler.class),
- new SegmentCacheManagerFactory(TestIndex.INDEX_IO, objectMapper),
+ SegmentCacheManagerFactory.createWithOwnedPool(TestIndex.INDEX_IO,
objectMapper),
objectMapper,
testUtils.getTestIndexIO(),
MapCache.create(1024),
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/worker/WorkerTaskManagerTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/worker/WorkerTaskManagerTest.java
index d77b83a75c2..3e35287894a 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/worker/WorkerTaskManagerTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/worker/WorkerTaskManagerTest.java
@@ -159,7 +159,7 @@ public class WorkerTaskManagerTest
null,
NoopJoinableFactory.INSTANCE,
null,
- new SegmentCacheManagerFactory(TestIndex.INDEX_IO, jsonMapper),
+
SegmentCacheManagerFactory.createWithOwnedPool(TestIndex.INDEX_IO, jsonMapper),
jsonMapper,
indexIO,
null,
diff --git
a/multi-stage-query/src/test/java/org/apache/druid/msq/test/CalciteMSQTestsHelper.java
b/multi-stage-query/src/test/java/org/apache/druid/msq/test/CalciteMSQTestsHelper.java
index 837a545db1d..15d74d025c4 100644
---
a/multi-stage-query/src/test/java/org/apache/druid/msq/test/CalciteMSQTestsHelper.java
+++
b/multi-stage-query/src/test/java/org/apache/druid/msq/test/CalciteMSQTestsHelper.java
@@ -94,7 +94,7 @@ public class CalciteMSQTestsHelper
@LazySingleton
public SegmentCacheManager provideSegmentCacheManager(ObjectMapper
testMapper, TempDirProducer tempDirProducer)
{
- return new SegmentCacheManagerFactory(TestIndex.INDEX_IO, testMapper)
+ return
SegmentCacheManagerFactory.createWithOwnedPool(TestIndex.INDEX_IO, testMapper)
.manufacturate(tempDirProducer.newTempFolder("test"), null, true);
}
diff --git
a/multi-stage-query/src/test/java/org/apache/druid/msq/test/MSQTestBase.java
b/multi-stage-query/src/test/java/org/apache/druid/msq/test/MSQTestBase.java
index 54fc466c618..02cb12d7444 100644
--- a/multi-stage-query/src/test/java/org/apache/druid/msq/test/MSQTestBase.java
+++ b/multi-stage-query/src/test/java/org/apache/druid/msq/test/MSQTestBase.java
@@ -473,7 +473,7 @@ public class MSQTestBase extends BaseCalciteQueryTest
indexIO = new IndexIO(objectMapper, ColumnConfig.DEFAULT);
segmentCacheManager =
- new SegmentCacheManagerFactory(indexIO,
objectMapper).manufacturate(newTempFolder("cacheManager"), null, true);
+ SegmentCacheManagerFactory.createWithOwnedPool(indexIO,
objectMapper).manufacturate(newTempFolder("cacheManager"), null, true);
testSegmentManager = new TestSegmentManager();
diff --git
a/processing/src/main/java/org/apache/druid/guice/annotations/EphemeralStorageLoading.java
b/processing/src/main/java/org/apache/druid/guice/annotations/EphemeralStorageLoading.java
new file mode 100644
index 00000000000..f2874236ee9
--- /dev/null
+++
b/processing/src/main/java/org/apache/druid/guice/annotations/EphemeralStorageLoading.java
@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.guice.annotations;
+
+import com.google.inject.BindingAnnotation;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * Binding annotation for the {@code StorageLoadingThreadPool} used by
ephemeral, per-task segment caches
+ */
+@Target({ElementType.FIELD, ElementType.PARAMETER, ElementType.METHOD})
+@Retention(RetentionPolicy.RUNTIME)
+@BindingAnnotation
+public @interface EphemeralStorageLoading
+{
+}
diff --git a/server/src/main/java/org/apache/druid/guice/StorageNodeModule.java
b/server/src/main/java/org/apache/druid/guice/StorageNodeModule.java
index 09b953204a4..24d4e360c1e 100644
--- a/server/src/main/java/org/apache/druid/guice/StorageNodeModule.java
+++ b/server/src/main/java/org/apache/druid/guice/StorageNodeModule.java
@@ -28,6 +28,7 @@ import com.google.inject.name.Named;
import com.google.inject.util.Providers;
import org.apache.druid.client.DruidServerConfig;
import org.apache.druid.discovery.DataNodeService;
+import org.apache.druid.guice.annotations.EphemeralStorageLoading;
import org.apache.druid.guice.annotations.Self;
import org.apache.druid.java.util.emitter.EmittingLogger;
import org.apache.druid.query.DruidProcessingConfig;
@@ -134,6 +135,20 @@ public class StorageNodeModule implements Module
return StorageLoadingThreadPool.createFromConfig(config);
}
+ /**
+ * Process-wide, always-virtual on-demand loading pool shared by the
ephemeral per-task segment caches built via
+ * {@code SegmentCacheManagerFactory} (tasks and MSQ workers).
+ */
+ @Provides
+ @LazySingleton
+ @EphemeralStorageLoading
+ public StorageLoadingThreadPool
getEphemeralStorageLoadingThreadPool(SegmentLoaderConfig config)
+ {
+ // Force virtual-storage mode: this pool serves per-task caches even when
the node itself is not in virtual-storage
+ // mode. Threads are created lazily, so an unused pool on a non-task
process is cheap.
+ return
StorageLoadingThreadPool.createFromConfig(config.withVirtualStorage(true));
+ }
+
@Provides
@LazySingleton
@Named(IS_SEGMENT_CACHE_CONFIGURED)
diff --git
a/server/src/main/java/org/apache/druid/segment/loading/SegmentLoaderConfig.java
b/server/src/main/java/org/apache/druid/segment/loading/SegmentLoaderConfig.java
index d90be578d3d..b9c9f0804fe 100644
---
a/server/src/main/java/org/apache/druid/segment/loading/SegmentLoaderConfig.java
+++
b/server/src/main/java/org/apache/druid/segment/loading/SegmentLoaderConfig.java
@@ -137,6 +137,34 @@ public class SegmentLoaderConfig
private long combinedMaxSize = 0;
+ public SegmentLoaderConfig()
+ {
+ }
+
+ private SegmentLoaderConfig(SegmentLoaderConfig other)
+ {
+ this.locations = other.locations;
+ this.lazyLoadOnStart = other.lazyLoadOnStart;
+ this.deleteOnRemove = other.deleteOnRemove;
+ this.dropSegmentDelayMillis = other.dropSegmentDelayMillis;
+ this.announceIntervalMillis = other.announceIntervalMillis;
+ this.numLoadingThreads = other.numLoadingThreads;
+ this.numBootstrapThreads = other.numBootstrapThreads;
+ this.numThreadsToLoadSegmentsIntoPageCacheOnDownload =
other.numThreadsToLoadSegmentsIntoPageCacheOnDownload;
+ this.numThreadsToLoadSegmentsIntoPageCacheOnBootstrap =
other.numThreadsToLoadSegmentsIntoPageCacheOnBootstrap;
+ this.infoDir = other.infoDir;
+ this.statusQueueMaxSize = other.statusQueueMaxSize;
+ this.virtualStorage = other.virtualStorage;
+ this.virtualStorageLoadThreads = other.virtualStorageLoadThreads;
+ this.virtualStorageUseVirtualThreads =
other.virtualStorageUseVirtualThreads;
+ this.virtualStorageIsEphemeral = other.virtualStorageIsEphemeral;
+ this.virtualStorageMetadataReservationEstimate =
other.virtualStorageMetadataReservationEstimate;
+ this.virtualStoragePartialDownloadsEnabled =
other.virtualStoragePartialDownloadsEnabled;
+ this.virtualStorageCoalesceGapBytes = other.virtualStorageCoalesceGapBytes;
+ this.virtualStorageMaxFetchRunBytes = other.virtualStorageMaxFetchRunBytes;
+ this.combinedMaxSize = other.combinedMaxSize;
+ }
+
public List<StorageLocationConfig> getLocations()
{
return locations;
@@ -263,6 +291,19 @@ public class SegmentLoaderConfig
return this;
}
+ /**
+ * Returns a copy of this config with {@link #virtualStorage} set to {@code
virtualStorage}. All other settings
+ * (notably {@link #getVirtualStorageLoadThreads()} and {@link
#isVirtualStorageUseVirtualThreads()}) are preserved.
+ * Used to derive an always-virtual config for the shared ephemeral
on-demand loading pool from a node config that
+ * may not itself run in virtual-storage mode.
+ */
+ public SegmentLoaderConfig withVirtualStorage(boolean virtualStorage)
+ {
+ final SegmentLoaderConfig copy = new SegmentLoaderConfig(this);
+ copy.virtualStorage = virtualStorage;
+ return copy;
+ }
+
/**
* Sets {@link #virtualStorageIsEphemeral}.
*/
diff --git
a/server/src/main/java/org/apache/druid/segment/loading/StorageLoadingThreadPool.java
b/server/src/main/java/org/apache/druid/segment/loading/StorageLoadingThreadPool.java
index 18668a51e2c..b76f347cb14 100644
---
a/server/src/main/java/org/apache/druid/segment/loading/StorageLoadingThreadPool.java
+++
b/server/src/main/java/org/apache/druid/segment/loading/StorageLoadingThreadPool.java
@@ -27,7 +27,6 @@ import com.google.common.util.concurrent.MoreExecutors;
import org.apache.druid.common.asyncresource.AsyncResource;
import org.apache.druid.common.asyncresource.AsyncResources;
import org.apache.druid.error.DruidException;
-import org.apache.druid.guice.StorageNodeModule;
import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.common.lifecycle.LifecycleStop;
import org.apache.druid.java.util.common.logger.Logger;
@@ -50,7 +49,7 @@ import java.util.concurrent.TimeUnit;
*/
public class StorageLoadingThreadPool
{
- private static final Logger log = new Logger(StorageNodeModule.class);
+ private static final Logger log = new Logger(StorageLoadingThreadPool.class);
private final ListeningExecutorService exec;
diff --git
a/server/src/test/java/org/apache/druid/guice/StorageNodeModuleTest.java
b/server/src/test/java/org/apache/druid/guice/StorageNodeModuleTest.java
index 0f63572ed73..06c01e1ad5a 100644
--- a/server/src/test/java/org/apache/druid/guice/StorageNodeModuleTest.java
+++ b/server/src/test/java/org/apache/druid/guice/StorageNodeModuleTest.java
@@ -27,10 +27,12 @@ import com.google.inject.name.Names;
import com.google.inject.util.Modules;
import org.apache.druid.discovery.DataNodeService;
import org.apache.druid.error.ExceptionMatcher;
+import org.apache.druid.guice.annotations.EphemeralStorageLoading;
import org.apache.druid.guice.annotations.Self;
import org.apache.druid.initialization.Initialization;
import org.apache.druid.query.DruidProcessingConfig;
import org.apache.druid.segment.loading.SegmentLoaderConfig;
+import org.apache.druid.segment.loading.StorageLoadingThreadPool;
import org.apache.druid.segment.loading.StorageLocationConfig;
import org.apache.druid.server.DruidNode;
import org.apache.druid.server.coordination.DruidServerMetadata;
@@ -167,6 +169,26 @@ public class StorageNodeModuleTest
.assertThrowsAndMatches(() ->
injector.getInstance(DruidServerMetadata.class));
}
+ @Test
+ public void testEphemeralStorageLoadingThreadPoolIsInjectedAndAvailable()
+ {
+ // The qualified ephemeral loading pool must resolve from the core
injector (StorageNodeModule is universal via
+ // CoreInjectorBuilder), so SegmentCacheManagerFactory's @Inject
constructor can be satisfied on every process.
+ // The provider forces virtual storage via config.withVirtualStorage(true)
regardless of the node's flag.
+ Mockito.when(segmentLoaderConfig.withVirtualStorage(true))
+ .thenReturn(new SegmentLoaderConfig().setVirtualStorage(true));
+
+ final StorageLoadingThreadPool pool = injector().getInstance(
+ Key.get(StorageLoadingThreadPool.class, EphemeralStorageLoading.class)
+ );
+ try {
+ Assert.assertTrue(pool.isAvailable());
+ }
+ finally {
+ pool.stop();
+ }
+ }
+
private Injector injector()
{
return makeInjector(INJECT_SERVER_TYPE_CONFIG);
diff --git
a/server/src/test/java/org/apache/druid/segment/loading/SegmentLoaderConfigTest.java
b/server/src/test/java/org/apache/druid/segment/loading/SegmentLoaderConfigTest.java
index 9c61e00e62e..a3db293b6ad 100644
---
a/server/src/test/java/org/apache/druid/segment/loading/SegmentLoaderConfigTest.java
+++
b/server/src/test/java/org/apache/druid/segment/loading/SegmentLoaderConfigTest.java
@@ -80,4 +80,22 @@ class SegmentLoaderConfigTest
);
Assertions.assertEquals(8388608L,
configured.getVirtualStorageMaxFetchRunBytes());
}
+
+ @Test
+ public void testWithVirtualStorageReturnsCopyAndDoesNotMutateOriginal()
+ {
+ final SegmentLoaderConfig original = new SegmentLoaderConfig();
+ Assertions.assertFalse(original.isVirtualStorage());
+
+ final SegmentLoaderConfig copy = original.withVirtualStorage(true);
+
+ // The copy has the flag flipped, while other settings are preserved.
+ Assertions.assertTrue(copy.isVirtualStorage());
+ Assertions.assertEquals(original.getVirtualStorageLoadThreads(),
copy.getVirtualStorageLoadThreads());
+ Assertions.assertEquals(original.isVirtualStorageUseVirtualThreads(),
copy.isVirtualStorageUseVirtualThreads());
+
+ // The original is untouched and the copy is a distinct instance.
+ Assertions.assertNotSame(original, copy);
+ Assertions.assertFalse(original.isVirtualStorage());
+ }
}
diff --git
a/server/src/test/java/org/apache/druid/segment/loading/StorageLoadingThreadPoolTest.java
b/server/src/test/java/org/apache/druid/segment/loading/StorageLoadingThreadPoolTest.java
new file mode 100644
index 00000000000..78c5fde240d
--- /dev/null
+++
b/server/src/test/java/org/apache/druid/segment/loading/StorageLoadingThreadPoolTest.java
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.segment.loading;
+
+import org.apache.druid.error.DruidException;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class StorageLoadingThreadPoolTest
+{
+ @Test
+ void testCreateFromConfigIsUnavailableWhenNotVirtualStorage()
+ {
+ Assertions.assertFalse(StorageLoadingThreadPool.createFromConfig(new
SegmentLoaderConfig()).isAvailable());
+ }
+
+ @Test
+ void testCreateFromConfigIsAvailableWhenVirtualStorage()
+ {
+ // The shared ephemeral pool is built this way:
createFromConfig(config.withVirtualStorage(true)).
+ final StorageLoadingThreadPool pool =
+ StorageLoadingThreadPool.createFromConfig(new
SegmentLoaderConfig().withVirtualStorage(true));
+ try {
+ Assertions.assertTrue(pool.isAvailable());
+ Assertions.assertNotNull(pool.getExecutorService());
+ }
+ finally {
+ pool.stop();
+ }
+ }
+
+ @Test
+ void testCreateFromConfigRejectsNonPositiveThreadCountInVirtualStorage()
+ {
+ final SegmentLoaderConfig config = new SegmentLoaderConfig()
+ {
+ @Override
+ public int getVirtualStorageLoadThreads()
+ {
+ return 0;
+ }
+ }.setVirtualStorage(true);
+ Assertions.assertThrows(DruidException.class, () ->
StorageLoadingThreadPool.createFromConfig(config));
+ }
+}
diff --git a/services/src/test/java/org/apache/druid/cli/CliPeonTest.java
b/services/src/test/java/org/apache/druid/cli/CliPeonTest.java
index f7f6eb7f60a..8d49c6b3747 100644
--- a/services/src/test/java/org/apache/druid/cli/CliPeonTest.java
+++ b/services/src/test/java/org/apache/druid/cli/CliPeonTest.java
@@ -105,7 +105,7 @@ public class CliPeonTest
private final ObjectMapper mapper = TestHelper.makeJsonMapper();
private final CompactionTask.Builder compactBuilder =
- new CompactionTask.Builder("test_ds", new
SegmentCacheManagerFactory(TestIndex.INDEX_IO, mapper))
+ new CompactionTask.Builder("test_ds",
SegmentCacheManagerFactory.createWithOwnedPool(TestIndex.INDEX_IO, mapper))
.id("compact_test_taskid")
.inputSpec(new CompactionIntervalSpec(Intervals.of("2020/2021"), null));
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
