You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by "clintropolis (via GitHub)" <gi...@apache.org> on 2026/07/06 21:33:08 UTC
[PR] fix: dedicated ephemeral storage loading pool for tasks (druid)
clintropolis opened a new pull request, #19658:
URL: https://github.com/apache/druid/pull/19658
### Description
Fixes an issue originally noticed in https://github.com/apache/druid/pull/19535#pullrequestreview-4445234363 though actually not from that PR, where the per task storage loading pool was not shutdown after task completion, resulting in leaked pools when used in an indexer process. The fix injects a new 'ephemeral' pool to re-use for tasks tied to the lifecycle.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] fix: dedicated ephemeral storage loading pool for tasks (druid)
Posted by "capistrant (via GitHub)" <gi...@apache.org>.
capistrant commented on code in PR #19658:
URL: https://github.com/apache/druid/pull/19658#discussion_r3545152423
##########
server/src/main/java/org/apache/druid/segment/loading/StorageLoadingThreadPool.java:
##########
Review Comment:
unrelated to the PR, but is this supposed to be `new Logger(StorageLoadingThreadPool.class);`?
##########
server/src/main/java/org/apache/druid/segment/loading/StorageLoadingThreadPool.java:
##########
@@ -63,49 +63,59 @@ public StorageLoadingThreadPool(
public static StorageLoadingThreadPool createFromConfig(final SegmentLoaderConfig config)
{
- final ListeningExecutorService exec;
+ return new StorageLoadingThreadPool(config.isVirtualStorage() ? createOnDemandLoadingExecutor(config) : null);
+ }
+
+ /**
+ * Build a pool configured for virtual-storage on-demand loading <b>regardless of</b>
+ * {@link SegmentLoaderConfig#isVirtualStorage()}. Used for the process-wide loading pool shared by ephemeral,
+ * per-task segment caches, whose host process may not itself run in virtual-storage mode. The executor is created
+ * eagerly but spawns no threads until work is submitted, so an unused pool is cheap.
+ *
+ * @see org.apache.druid.guice.annotations.EphemeralStorageLoading
+ */
+ public static StorageLoadingThreadPool createForEphemeral(final SegmentLoaderConfig config)
+ {
+ return new StorageLoadingThreadPool(createOnDemandLoadingExecutor(config));
+ }
- if (config.isVirtualStorage()) {
- if (config.getVirtualStorageLoadThreads() <= 0) {
- throw DruidException.forPersona(DruidException.Persona.OPERATOR)
- .ofCategory(DruidException.Category.INVALID_INPUT)
- .build(
- "virtualStorageLoadThreads must be greater than 0, got [%d]",
- config.getVirtualStorageLoadThreads()
- );
- }
- if (config.isVirtualStorageUseVirtualThreads()) {
- log.info(
- "Using virtual storage mode with virtual threads - max concurrent on demand loads: [%d].",
- config.getVirtualStorageLoadThreads()
- );
- exec = new PermitBoundedListeningExecutorService(
- MoreExecutors.listeningDecorator(
- Executors.newThreadPerTaskExecutor(
- Thread.ofVirtual()
- .name("VirtualStorageOnDemandLoadingThread-", 0)
- .factory()
- )
- ),
- new Semaphore(config.getVirtualStorageLoadThreads())
- );
- } else {
- log.info(
- "Using virtual storage mode with fixed platform thread pool - on demand load threads: [%d].",
- config.getVirtualStorageLoadThreads()
- );
- exec = MoreExecutors.listeningDecorator(
- Executors.newFixedThreadPool(
- config.getVirtualStorageLoadThreads(),
- Execs.makeThreadFactory("VirtualStorageOnDemandLoadingThread-%s")
- )
- );
- }
+ private static ListeningExecutorService createOnDemandLoadingExecutor(final SegmentLoaderConfig config)
Review Comment:
Consider value of adding a "purpose" parameter that can enrich the logs based on if it is the ephemeral create or the create from config. would help avoid confusion for an operator who doesn't have process level virtual storage flipped on but still sees logs for it due to ephemeral create. not blocking
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] fix: dedicated ephemeral storage loading pool for tasks (druid)
Posted by "gianm (via GitHub)" <gi...@apache.org>.
gianm commented on code in PR #19658:
URL: https://github.com/apache/druid/pull/19658#discussion_r3547150331
##########
indexing-service/src/main/java/org/apache/druid/indexing/common/SegmentCacheManagerFactory.java:
##########
@@ -43,26 +46,45 @@ public class SegmentCacheManagerFactory
{
private final IndexIO indexIO;
private final ObjectMapper jsonMapper;
+ private final StorageLoadingThreadPool loadingThreadPool;
@Inject
public SegmentCacheManagerFactory(
IndexIO indexIO,
- @Json ObjectMapper mapper
+ @Json ObjectMapper mapper,
+ @EphemeralStorageLoading StorageLoadingThreadPool loadingThreadPool
Review Comment:
Hmm. The `SegmentCacheManagerFactory` is injected into `DruidInputSource`, which will cause the pool to be created anywhere that a `DruidInputSource` is instantiated. If used in task specs, that would be Overlord, MM, Peon, Indexer, and possibly even Broker. I don't think we want the `EphemeralStorageLoading` pool to exist in all these places. It won't even be used by the `DruidInputSource` anyway, since `manufacturate` is called with `virtualStorage = false`.
Maybe this can be fixed by injecting `IndexIO` and `@Json ObjectMapper` directly into `DruidInputSource`, rather than injecting `SegmentCacheManagerFactory`?
##########
server/src/main/java/org/apache/druid/segment/loading/StorageLoadingThreadPool.java:
##########
@@ -63,49 +63,59 @@ public StorageLoadingThreadPool(
public static StorageLoadingThreadPool createFromConfig(final SegmentLoaderConfig config)
{
- final ListeningExecutorService exec;
+ return new StorageLoadingThreadPool(config.isVirtualStorage() ? createOnDemandLoadingExecutor(config) : null);
+ }
+
+ /**
+ * Build a pool configured for virtual-storage on-demand loading <b>regardless of</b>
+ * {@link SegmentLoaderConfig#isVirtualStorage()}. Used for the process-wide loading pool shared by ephemeral,
+ * per-task segment caches, whose host process may not itself run in virtual-storage mode. The executor is created
+ * eagerly but spawns no threads until work is submitted, so an unused pool is cheap.
+ *
+ * @see org.apache.druid.guice.annotations.EphemeralStorageLoading
+ */
+ public static StorageLoadingThreadPool createForEphemeral(final SegmentLoaderConfig config)
Review Comment:
IMO it would be cleaner to remove this method, add `withVirtualStorage` to `SegmentLoaderConfig`, and have the Guice module call `createFromConfig` with a config adjusted to have `virtualStorage = true`. (Alternatively: add `copy` to `SegmentLoaderConfig` and then call `setVirtualStorage` on the copy.)
This will simplify the logic in `StorageLoadingThreadPool` and make it more obvious what is happening. IMO, it is not obvious what "createForEphemeral" means, but it _would_ be obvious what `createFromConfig(config.withVirtualStorage(true))` means.
##########
indexing-service/src/main/java/org/apache/druid/indexing/common/SegmentCacheManagerFactory.java:
##########
@@ -43,26 +46,45 @@ public class SegmentCacheManagerFactory
{
private final IndexIO indexIO;
private final ObjectMapper jsonMapper;
+ private final StorageLoadingThreadPool loadingThreadPool;
@Inject
public SegmentCacheManagerFactory(
IndexIO indexIO,
- @Json ObjectMapper mapper
+ @Json ObjectMapper mapper,
+ @EphemeralStorageLoading StorageLoadingThreadPool loadingThreadPool
)
{
this.indexIO = indexIO;
this.jsonMapper = mapper;
+ this.loadingThreadPool = loadingThreadPool;
}
/**
- * Creates a new {@link SegmentCacheManager} backed by a new storage location in {@code storageDir}, and a new
- * loading thread pool of default size.
+ * Convenience constructor for tests and manual construction. Builds a private, always-virtual loading pool of default
+ * size rather than sharing the process-wide {@link EphemeralStorageLoading} pool. The pool is not lifecycle-managed;
+ * this is fine for short-lived test JVMs.
+ */
+ @VisibleForTesting
+ public SegmentCacheManagerFactory(
Review Comment:
I believe that `@VisibleForTesting` is more for things that are used in both production and tests, where the visibility is broader than necessary for production, due to the requirements of the tests. This constructor seems like something that is test-only. I would suggest making it a static factory method instead, like `SegmentCacheManagerFactory.createWithOwnedPool`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] fix: dedicated ephemeral storage loading pool for tasks (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on code in PR #19658:
URL: https://github.com/apache/druid/pull/19658#discussion_r3595313300
##########
server/src/main/java/org/apache/druid/segment/loading/SegmentLoaderConfig.java:
##########
@@ -137,6 +137,32 @@ 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.combinedMaxSize = other.combinedMaxSize;
Review Comment:
[P2] Preserve all virtual-storage fetch settings in the copy
The new copy constructor ends here without copying `virtualStorageCoalesceGapBytes` or `virtualStorageMaxFetchRunBytes`, even though `withVirtualStorage` promises every other setting is preserved. A non-default configured value is therefore silently reset to the class default in the returned config. Copy both fields and extend the test with non-default values.
##########
indexing-service/src/main/java/org/apache/druid/indexing/common/SegmentCacheManagerFactory.java:
##########
@@ -77,11 +107,10 @@ public SegmentCacheManager manufacturate(File storageDir, Long maxSize, boolean
.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(),
Review Comment:
[P2] Keep the pool alive until task shutdown completes
On the Indexer, this provider is first resolved from a running task, after the NORMAL-scoped `WorkerTaskManager` has been registered. Lifecycle shutdown reverses NORMAL registration order, so `StorageLoadingThreadPool.stop()` runs first and calls `shutdownNow()` before `WorkerTaskManager` invokes `ThreadingTaskRunner.stop()` to gracefully stop active tasks. Any task still doing or scheduling on-demand loads is interrupted or rejected during cleanup. Register this pool in an earlier lifecycle stage, or otherwise before the worker manager, while retaining lazy provisioning.
##########
indexing-service/src/main/java/org/apache/druid/indexing/common/SegmentCacheManagerFactory.java:
##########
@@ -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)
Review Comment:
[P2] Retain the existing two-argument constructor
The follow-up replaces the existing public `SegmentCacheManagerFactory(IndexIO, ObjectMapper)` constructor with this static factory and updates only repository call sites. External extensions or tests compiled against that constructor will now fail to link or compile on upgrade, despite equivalent behavior remaining available. Keep a delegating, possibly deprecated, two-argument constructor unless this compatibility break is intentional.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] fix: dedicated ephemeral storage loading pool for tasks (druid)
Posted by "gianm (via GitHub)" <gi...@apache.org>.
gianm merged PR #19658:
URL: https://github.com/apache/druid/pull/19658
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] fix: dedicated ephemeral storage loading pool for tasks (druid)
Posted by "clintropolis (via GitHub)" <gi...@apache.org>.
clintropolis commented on code in PR #19658:
URL: https://github.com/apache/druid/pull/19658#discussion_r3606327515
##########
indexing-service/src/main/java/org/apache/druid/indexing/common/SegmentCacheManagerFactory.java:
##########
@@ -77,11 +107,10 @@ public SegmentCacheManager manufacturate(File storageDir, Long maxSize, boolean
.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(),
Review Comment:
i just removed this from the lifecycle and marked it lazy singleton instead, which I think should be ok as I would expect nothing to be running on the pool by the time of shutdown since all of the threads running things that could be using it will have been stopped by the time of process exit.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] fix: dedicated ephemeral storage loading pool for tasks (druid)
Posted by "gianm (via GitHub)" <gi...@apache.org>.
gianm commented on code in PR #19658:
URL: https://github.com/apache/druid/pull/19658#discussion_r3596333340
##########
server/src/main/java/org/apache/druid/segment/loading/SegmentLoaderConfig.java:
##########
@@ -137,6 +137,32 @@ 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.combinedMaxSize = other.combinedMaxSize;
Review Comment:
+1, `virtualStorageCoalesceGapBytes` and `virtualStorageMaxFetchRunBytes` are apparently missing.
##########
indexing-service/src/main/java/org/apache/druid/indexing/common/SegmentCacheManagerFactory.java:
##########
@@ -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)
Review Comment:
IMO it's OK (and preferred) to remove the 2-arg constructor.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org