This is an automated email from the ASF dual-hosted git repository.
clintropolis 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 f0cbad8f0f0 feat: support partial loads for tasks (#19739)
f0cbad8f0f0 is described below
commit f0cbad8f0f05e2b9e728782761c8f236b303d1f6
Author: Clint Wylie <[email protected]>
AuthorDate: Tue Jul 28 15:48:54 2026 -0700
feat: support partial loads for tasks (#19739)
---
.../msq/EmbeddedMSQPartialDownloadsTest.java | 204 +++++++++
.../MultipleKubernetesTaskRunnerFactoryTest.java | 2 +-
.../common/SegmentCacheManagerFactory.java | 49 ++-
.../druid/indexing/common/TaskToolboxFactory.java | 12 +-
.../druid/indexing/common/config/TaskConfig.java | 27 +-
.../apache/druid/indexing/common/task/Tasks.java | 1 +
.../druid/indexing/input/DruidInputSource.java | 2 +-
.../common/SegmentCacheManagerFactoryTest.java | 116 ++++-
.../druid/indexing/common/TaskToolboxTest.java | 4 +
.../indexing/common/config/TaskConfigBuilder.java | 10 +-
.../indexing/common/config/TaskConfigTest.java | 45 ++
.../common/task/CompactionTaskRunBase.java | 53 ++-
.../druid/indexing/common/task/IndexTaskTest.java | 13 +-
.../indexing/common/task/IngestionTestBase.java | 2 +-
.../AbstractMultiPhaseParallelIndexingTest.java | 2 +-
.../druid/msq/indexing/IndexerWorkerContext.java | 8 +-
.../druid/msq/util/MultiStageQueryContext.java | 15 +
.../msq/input/RegularLoadableSegmentTest.java | 17 +-
.../druid/msq/test/CalciteMSQTestsHelper.java | 2 +-
.../org/apache/druid/msq/test/MSQTestBase.java | 3 +-
.../druid/msq/util/MultiStageQueryContextTest.java | 24 +
.../org/apache/druid/guice/StorageNodeModule.java | 2 +-
.../druid/segment/loading/SegmentLoaderConfig.java | 487 ++++++++++++++++-----
.../segment/loading/SegmentLocalCacheManager.java | 6 +
.../apache/druid/client/DruidServerConfigTest.java | 13 +-
.../apache/druid/guice/StorageNodeModuleTest.java | 6 +-
.../PermitLimitedSegmentRangeReaderTest.java | 9 +-
.../segment/loading/SegmentLoaderConfigTest.java | 68 ++-
.../SegmentLocalCacheManagerConcurrencyTest.java | 50 +--
...SegmentLocalCacheManagerPartialAcquireTest.java | 38 +-
.../SegmentLocalCacheManagerPartialDropTest.java | 2 +-
...egmentLocalCacheManagerPartialRuleLoadTest.java | 9 +-
.../loading/SegmentLocalCacheManagerTest.java | 246 +++--------
.../loading/StorageLoadingThreadPoolTest.java | 46 +-
.../appenderator/StreamAppenderatorTester.java | 11 +-
.../apache/druid/server/SegmentManagerTest.java | 48 +-
.../apache/druid/server/StatusResourceTest.java | 6 +-
.../SegmentCacheBootstrapperCacheTest.java | 23 +-
.../coordination/SegmentCacheBootstrapperTest.java | 35 +-
.../coordination/SegmentLoadDropHandlerTest.java | 83 +---
...egmentManagerBroadcastJoinIndexedTableTest.java | 20 +-
.../SegmentManagerThreadSafetyTest.java | 20 +-
.../server/metrics/SegmentStatsMonitorTest.java | 2 +-
43 files changed, 1135 insertions(+), 706 deletions(-)
diff --git
a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/msq/EmbeddedMSQPartialDownloadsTest.java
b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/msq/EmbeddedMSQPartialDownloadsTest.java
new file mode 100644
index 00000000000..af3581b6563
--- /dev/null
+++
b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/msq/EmbeddedMSQPartialDownloadsTest.java
@@ -0,0 +1,204 @@
+/*
+ * 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.testing.embedded.msq;
+
+import org.apache.druid.common.utils.IdUtils;
+import org.apache.druid.data.input.impl.LocalInputSource;
+import org.apache.druid.indexer.TaskState;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.msq.indexing.report.MSQTaskReportPayload;
+import org.apache.druid.msq.util.MultiStageQueryContext;
+import org.apache.druid.query.DruidProcessingConfigTest;
+import org.apache.druid.query.http.SqlTaskStatus;
+import org.apache.druid.server.metrics.LatchableEmitter;
+import org.apache.druid.server.metrics.StorageMonitor;
+import org.apache.druid.sql.calcite.planner.Calcites;
+import org.apache.druid.testing.embedded.EmbeddedBroker;
+import org.apache.druid.testing.embedded.EmbeddedCoordinator;
+import org.apache.druid.testing.embedded.EmbeddedDruidCluster;
+import org.apache.druid.testing.embedded.EmbeddedHistorical;
+import org.apache.druid.testing.embedded.EmbeddedIndexer;
+import org.apache.druid.testing.embedded.EmbeddedOverlord;
+import org.apache.druid.testing.embedded.EmbeddedRouter;
+import org.apache.druid.testing.embedded.junit5.EmbeddedClusterTestBase;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.util.Collections;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * End-to-end wiring test for on-demand partial segment downloads on MSQ
worker tasks. Partial downloads default to
+ * enabled for tasks (the {@code
druid.indexer.task.virtualStoragePartialDownloadsEnabled} {@code TaskConfig}
default is
+ * true), so run 1 sets no property or context and verifies the default
engages the partial path. Run 2 verifies the
+ * per-query {@link
MultiStageQueryContext#CTX_VIRTUAL_STORAGE_PARTIAL_DOWNLOADS} override can turn
it back off.
+ * <p>
+ * The observable signal is the indexer's {@link
StorageMonitor#VSF_READ_COUNT} metric: the partial path performs
+ * on-demand deep-storage range reads (so the count is positive), while the
full-download path does not (the count
+ * stays zero because {@code virtualStoragePartialDownloadsEnabled=false}
makes the worker mount segments as complete
+ * cache entries and download them in full instead of via range reads).
+ */
+class EmbeddedMSQPartialDownloadsTest extends EmbeddedClusterTestBase
+{
+ // A few times the storage monitor's PT1s emission period, so a single
missed tick doesn't falsely read as "idle".
+ private static final long MONITOR_QUIESCE_TIMEOUT_MILLIS = 3_000L;
+
+ private static final String SELECT_SQL = "SELECT channel, COUNT(*) AS c FROM
\"%s\" GROUP BY channel";
+
+ private final EmbeddedBroker broker = new EmbeddedBroker();
+ private final EmbeddedIndexer indexer = new EmbeddedIndexer();
+ private final EmbeddedOverlord overlord = new EmbeddedOverlord();
+ private final EmbeddedHistorical historical = new EmbeddedHistorical();
+ private final EmbeddedCoordinator coordinator = new EmbeddedCoordinator();
+ private final EmbeddedRouter router = new EmbeddedRouter();
+
+ private EmbeddedMSQApis msqApis;
+
+ @Override
+ public EmbeddedDruidCluster createCluster()
+ {
+ indexer.setServerMemory(400_000_000)
+ .addProperty("druid.worker.capacity", "4")
+ .addProperty("druid.processing.numThreads", "3");
+
+ broker.setServerMemory(200_000_000);
+
+ coordinator.addProperty("druid.manager.segments.useIncrementalCache",
"always");
+ overlord.addProperty("druid.manager.segments.useIncrementalCache",
"always")
+ .addProperty("druid.manager.segments.pollDuration", "PT0.1s");
+
+ return EmbeddedDruidCluster
+ .withEmbeddedDerbyAndZookeeper()
+ .useLatchableEmitter()
+ .useDefaultTimeoutForLatchableEmitter(20)
+ // Partial downloads require unzipped, V10 segments (range-readable
via LocalLoadSpec's directory reader).
+ .addCommonProperty("druid.storage.zip", "false")
+ .addCommonProperty("druid.indexer.task.buildV10", "true")
+ .addCommonProperty("druid.monitoring.emissionPeriod", "PT1s")
+ .addServer(coordinator)
+ .addServer(overlord)
+ .addServer(indexer)
+ .addServer(historical)
+ .addServer(broker)
+ .addServer(router);
+ }
+
+ @BeforeAll
+ void loadData() throws IOException
+ {
+ msqApis = new EmbeddedMSQApis(cluster, overlord);
+ dataSource = "wiki-" + IdUtils.getRandomId();
+ loadWikiData();
+ }
+
+ @Override
+ protected void refreshDatasourceName()
+ {
+ // Keep the datasource stable: it is ingested once in loadData() before
all tests.
+ }
+
+ @Test
+ void testPartialDownloadsWiredViaRuntimePropertyAndQueryContext()
+ {
+ final LatchableEmitter emitter = indexer.latchableEmitter();
+
+ // Run 1: no context override, so the query uses the indexer's TaskConfig
default (partial enabled). The worker
+ // reads the input segment's columns on demand, which shows up as
deep-storage range reads.
+ emitter.awaitMetricQuiescent(StorageMonitor.VSF_READ_COUNT,
MONITOR_QUIESCE_TIMEOUT_MILLIS);
+ emitter.flush();
+ runMsqSelect(Collections.emptyMap());
+ // The worker's StorageMonitor only emits while the (ephemeral) task is
alive, so the VSF_READ_COUNT events land
+ // during runMsqSelect. waitForEventAggregate (unlike waitForNextEvent)
also considers events already emitted since
+ // the last flush, so it sees them; it fails the test (timeout) if no
on-demand range reads ever happened.
+ emitter.waitForEventAggregate(
+ matcher -> matcher.hasMetricName(StorageMonitor.VSF_READ_COUNT),
+ aggregate -> aggregate.hasSumAtLeast(1L)
+ );
+
+ // Run 2: the query context disables partial downloads, so the worker
downloads segments in full and performs no
+ // range reads. The flag is passed as the string "false" (as it can arrive
from a client): it is copied into the
+ // worker task context and must be parsed, not unboxed, when building the
toolbox and worker context.
+ emitter.awaitMetricQuiescent(StorageMonitor.VSF_READ_COUNT,
MONITOR_QUIESCE_TIMEOUT_MILLIS);
+ emitter.flush();
+
runMsqSelect(Map.of(MultiStageQueryContext.CTX_VIRTUAL_STORAGE_PARTIAL_DOWNLOADS,
"false"));
+ emitter.awaitMetricQuiescent(StorageMonitor.VSF_READ_COUNT,
MONITOR_QUIESCE_TIMEOUT_MILLIS);
+ Assertions.assertEquals(
+ 0L,
+ emitter.getMetricEventLongSum(StorageMonitor.VSF_READ_COUNT),
+ "with partial downloads disabled via query context, the worker should
download in full (no range reads)"
+ );
+ }
+
+ private void runMsqSelect(Map<String, Object> queryContext)
+ {
+ final SqlTaskStatus status = msqApis.submitTaskSql(queryContext,
SELECT_SQL, dataSource);
+ cluster.callApi().waitForTaskToSucceed(status.getTaskId(), overlord);
+ }
+
+ private void loadWikiData() throws IOException
+ {
+ final File tmpDir = cluster.getTestFolder().newFolder();
+ final File wikiFile = new File(tmpDir, "wiki.gz");
+ try (var in =
DruidProcessingConfigTest.class.getResourceAsStream("/wikipedia/wikiticker-2015-09-12-sampled.json.gz"))
{
+ Files.copy(
+ Objects.requireNonNull(in, "wikiticker sample resource not found on
the test classpath"),
+ wikiFile.toPath()
+ );
+ }
+
+ final String sql = StringUtils.format(
+ "SET waitUntilSegmentsLoad = TRUE;\n"
+ + "REPLACE INTO \"%s\" OVERWRITE ALL\n"
+ + "SELECT\n"
+ + " TIME_PARSE(\"time\") AS __time,\n"
+ + " channel,\n"
+ + " countryName,\n"
+ + " page,\n"
+ + " \"user\",\n"
+ + " added,\n"
+ + " deleted,\n"
+ + " delta\n"
+ + "FROM TABLE(\n"
+ + " EXTERN(\n"
+ + " %s,\n"
+ + " '{\"type\":\"json\"}',\n"
+ + "
'[{\"name\":\"channel\",\"type\":\"string\"},{\"name\":\"time\",\"type\":\"string\"},{\"name\":\"page\",\"type\":\"string\"},{\"name\":\"added\",\"type\":\"long\"},{\"name\":\"user\",\"type\":\"string\"},{\"name\":\"delta\",\"type\":\"long\"},{\"name\":\"deleted\",\"type\":\"long\"},{\"name\":\"countryName\",\"type\":\"string\"}]'\n"
+ + " )\n"
+ + " )\n"
+ + "PARTITIONED BY DAY\n"
+ + "CLUSTERED BY channel",
+ dataSource,
+ Calcites.escapeStringLiteral(
+ broker.bindings()
+ .jsonMapper()
+ .writeValueAsString(new LocalInputSource(null, null,
Collections.singletonList(wikiFile), null))
+ )
+ );
+
+ final MSQTaskReportPayload payload = msqApis.runTaskSqlAndGetReport(sql);
+ Assertions.assertEquals(TaskState.SUCCESS,
payload.getStatus().getStatus());
+ Assertions.assertNull(payload.getStatus().getErrorReport());
+ }
+}
diff --git
a/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/MultipleKubernetesTaskRunnerFactoryTest.java
b/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/MultipleKubernetesTaskRunnerFactoryTest.java
index 7d0c3f20a6b..160c3eabd9e 100644
---
a/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/MultipleKubernetesTaskRunnerFactoryTest.java
+++
b/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/MultipleKubernetesTaskRunnerFactoryTest.java
@@ -209,7 +209,7 @@ public class MultipleKubernetesTaskRunnerFactoryTest
new NoopServiceEmitter(),
() -> null,
configManager,
- new TaskConfig(null, null, false, null, null, null, false, null,
false, null, false),
+ new TaskConfig(null, null, false, null, null, null, false, null,
false, null, false, null),
new StartupLoggingConfig(),
new DruidNode("test-overlord", "localhost", false, 8080, null, true,
false),
new DruidKubernetesVertxHttpClientFactory(new
DruidKubernetesVertxHttpClientConfig(), objectMapper)
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 402e4be10c5..9897338f9fa 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
@@ -37,7 +37,6 @@ import org.apache.druid.segment.loading.StorageLocationConfig;
import org.apache.druid.timeline.DataSegment;
import java.io.File;
-import java.util.Collections;
import java.util.List;
/**
@@ -47,6 +46,11 @@ public class SegmentCacheManagerFactory
{
private final IndexIO indexIO;
private final ObjectMapper jsonMapper;
+ /**
+ * The injected {@code druid.segmentCache} config. Per-task caches are
derived from it (see {@link #manufacturate})
+ * so operator-tuned virtual-storage settings.
+ */
+ private final SegmentLoaderConfig segmentLoaderConfig;
/**
* 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
@@ -59,11 +63,13 @@ public class SegmentCacheManagerFactory
public SegmentCacheManagerFactory(
IndexIO indexIO,
@Json ObjectMapper mapper,
+ SegmentLoaderConfig segmentLoaderConfig,
@EphemeralStorageLoading Provider<StorageLoadingThreadPool>
loadingThreadPoolProvider
)
{
this.indexIO = indexIO;
this.jsonMapper = mapper;
+ this.segmentLoaderConfig = segmentLoaderConfig;
this.loadingThreadPoolProvider = loadingThreadPoolProvider;
}
@@ -75,10 +81,12 @@ public class SegmentCacheManagerFactory
*/
public static SegmentCacheManagerFactory createWithOwnedPool(IndexIO
indexIO, ObjectMapper mapper)
{
+ final SegmentLoaderConfig defaults = SegmentLoaderConfig.builder().build();
return new SegmentCacheManagerFactory(
indexIO,
mapper,
- Providers.of(StorageLoadingThreadPool.createFromConfig(new
SegmentLoaderConfig().setVirtualStorage(true)))
+ defaults,
+
Providers.of(StorageLoadingThreadPool.createFromConfig(defaults.toEphemeralVirtualStorage()))
);
}
@@ -88,24 +96,41 @@ public class SegmentCacheManagerFactory
* {@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, AcquireMode)},
- * and segment files are deleted as soon as all holds
are closed.
+ * @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, AcquireMode)}, and segment
+ * files are deleted as soon as all holds are
closed.
+ * @param partialDownloadsEnabled when true (and {@code virtualStorage} is
true), partial-eligible segments are read
+ * via on-demand column downloads rather than
downloaded in full up front. Has no
+ * effect when {@code virtualStorage} is
false.
*/
- public SegmentCacheManager manufacturate(File storageDir, Long maxSize,
boolean virtualStorage)
+ public SegmentCacheManager manufacturate(
+ File storageDir,
+ Long maxSize,
+ boolean virtualStorage,
+ boolean partialDownloadsEnabled
+ )
{
final StorageLocationConfig locationConfig = new StorageLocationConfig(
storageDir,
maxSize != null ? maxSize : Long.MAX_VALUE,
null
);
+ // For virtual storage, derive an ephemeral virtual-storage cache config
from the injected node config: this keeps
+ // its virtual-storage tuning while dropping the classic on-disk-cache
settings that don't apply. A non-virtual
+ // (eager) per-task cache needs none of that tuning and must NOT inherit
the node's transient settings, so it
+ // starts from fresh defaults instead. Both then get the per-task location
and mode flags.
+ final SegmentLoaderConfig.Builder loaderConfigBuilder =
+ virtualStorage ?
segmentLoaderConfig.toEphemeralVirtualStorage().toBuilder() :
SegmentLoaderConfig.builder();
final SegmentLoaderConfig loaderConfig =
- new SegmentLoaderConfig()
- .setLocations(Collections.singletonList(locationConfig))
- .setVirtualStorage(virtualStorage)
- .setVirtualStorageIsEphemeral(virtualStorage);
+ loaderConfigBuilder
+ .locations(locationConfig)
+ .virtualStorage(virtualStorage)
+ .virtualStorageIsEphemeral(virtualStorage)
+ .virtualStoragePartialDownloadsEnabled(partialDownloadsEnabled)
+ .build();
final List<StorageLocation> storageLocations =
loaderConfig.toStorageLocations();
return new SegmentLocalCacheManager(
storageLocations,
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/common/TaskToolboxFactory.java
b/indexing-service/src/main/java/org/apache/druid/indexing/common/TaskToolboxFactory.java
index 6ee32438d49..655e3399e2a 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/common/TaskToolboxFactory.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/common/TaskToolboxFactory.java
@@ -45,6 +45,7 @@ import
org.apache.druid.indexing.worker.shuffle.IntermediaryDataManager;
import org.apache.druid.java.util.emitter.service.ServiceEmitter;
import org.apache.druid.java.util.metrics.MonitorScheduler;
import org.apache.druid.query.DruidProcessingConfig;
+import org.apache.druid.query.QueryContext;
import org.apache.druid.query.QueryProcessingPool;
import org.apache.druid.query.QueryRunnerFactoryConglomerate;
import org.apache.druid.query.policy.PolicyEnforcer;
@@ -247,7 +248,16 @@ public class TaskToolboxFactory
.queryProcessingPool(queryProcessingPool)
.joinableFactory(joinableFactory)
.monitorSchedulerProvider(monitorSchedulerProvider)
-
.segmentCacheManager(segmentCacheManagerFactory.manufacturate(taskWorkDir,
null, true))
+ .segmentCacheManager(segmentCacheManagerFactory.manufacturate(
+ taskWorkDir,
+ null,
+ true,
+ // getBoolean (not the raw getContextValue) so a string
"true"/"false" is parsed
+ QueryContext.of(task.getContext()).getBoolean(
+ Tasks.VIRTUAL_STORAGE_PARTIAL_DOWNLOADS_KEY,
+ config.isVirtualStoragePartialDownloadsEnabled()
+ )
+ ))
.jsonMapper(jsonMapper)
.taskWorkDir(taskWorkDir)
.indexIO(indexIO)
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/common/config/TaskConfig.java
b/indexing-service/src/main/java/org/apache/druid/indexing/common/config/TaskConfig.java
index d043fa7c825..1b65b9f6843 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/common/config/TaskConfig.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/common/config/TaskConfig.java
@@ -45,6 +45,7 @@ public class TaskConfig implements TaskDirectory
private static final Period DEFAULT_GRACEFUL_SHUTDOWN_TIMEOUT = new
Period("PT5M");
private static final boolean DEFAULT_STORE_EMPTY_COLUMNS = true;
private static final long DEFAULT_TMP_STORAGE_BYTES_PER_TASK = -1;
+ private static final boolean
DEFAULT_VIRTUAL_STORAGE_PARTIAL_DOWNLOADS_ENABLED = true;
@JsonProperty
private final String baseDir;
@@ -79,6 +80,9 @@ public class TaskConfig implements TaskDirectory
@JsonProperty
private final boolean buildV10;
+ @JsonProperty
+ private final boolean virtualStoragePartialDownloadsEnabled;
+
@JsonCreator
public TaskConfig(
@JsonProperty("baseDir") String baseDir,
@@ -91,7 +95,8 @@ public class TaskConfig implements TaskDirectory
@JsonProperty("storeEmptyColumns") @Nullable Boolean storeEmptyColumns,
@JsonProperty("encapsulatedTask") boolean enableTaskLevelLogPush,
@JsonProperty("tmpStorageBytesPerTask") @Nullable Long
tmpStorageBytesPerTask,
- @JsonProperty("buildV10") boolean buildV10
+ @JsonProperty("buildV10") boolean buildV10,
+ @JsonProperty("virtualStoragePartialDownloadsEnabled") @Nullable Boolean
virtualStoragePartialDownloadsEnabled
)
{
this.baseDir = Configs.valueOrDefault(baseDir,
System.getProperty("java.io.tmpdir"));
@@ -118,6 +123,10 @@ public class TaskConfig implements TaskDirectory
this.storeEmptyColumns = Configs.valueOrDefault(storeEmptyColumns,
DEFAULT_STORE_EMPTY_COLUMNS);
this.tmpStorageBytesPerTask =
Configs.valueOrDefault(tmpStorageBytesPerTask,
DEFAULT_TMP_STORAGE_BYTES_PER_TASK);
this.buildV10 = buildV10;
+ this.virtualStoragePartialDownloadsEnabled = Configs.valueOrDefault(
+ virtualStoragePartialDownloadsEnabled,
+ DEFAULT_VIRTUAL_STORAGE_PARTIAL_DOWNLOADS_ENABLED
+ );
}
private TaskConfig(
@@ -131,7 +140,8 @@ public class TaskConfig implements TaskDirectory
boolean storeEmptyColumns,
boolean encapsulatedTask,
long tmpStorageBytesPerTask,
- boolean buildV10
+ boolean buildV10,
+ boolean virtualStoragePartialDownloadsEnabled
)
{
this.baseDir = baseDir;
@@ -145,6 +155,7 @@ public class TaskConfig implements TaskDirectory
this.encapsulatedTask = encapsulatedTask;
this.tmpStorageBytesPerTask = tmpStorageBytesPerTask;
this.buildV10 = buildV10;
+ this.virtualStoragePartialDownloadsEnabled =
virtualStoragePartialDownloadsEnabled;
}
@JsonProperty
@@ -243,6 +254,12 @@ public class TaskConfig implements TaskDirectory
return buildV10;
}
+ @JsonProperty
+ public boolean isVirtualStoragePartialDownloadsEnabled()
+ {
+ return virtualStoragePartialDownloadsEnabled;
+ }
+
private String defaultDir(@Nullable String configParameter, final String
defaultVal)
{
if (configParameter == null) {
@@ -265,7 +282,8 @@ public class TaskConfig implements TaskDirectory
storeEmptyColumns,
encapsulatedTask,
tmpStorageBytesPerTask,
- buildV10
+ buildV10,
+ virtualStoragePartialDownloadsEnabled
);
}
@@ -282,7 +300,8 @@ public class TaskConfig implements TaskDirectory
storeEmptyColumns,
encapsulatedTask,
tmpStorageBytesPerTask,
- buildV10
+ buildV10,
+ virtualStoragePartialDownloadsEnabled
);
}
}
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/Tasks.java
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/Tasks.java
index 7f64a6cbad0..5e09198828e 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/Tasks.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/Tasks.java
@@ -55,6 +55,7 @@ public class Tasks
public static final String SUB_TASK_TIMEOUT_KEY = "subTaskTimeoutMillis";
public static final String FORCE_TIME_CHUNK_LOCK_KEY = "forceTimeChunkLock";
public static final String STORE_EMPTY_COLUMNS_KEY = "storeEmptyColumns";
+ public static final String VIRTUAL_STORAGE_PARTIAL_DOWNLOADS_KEY =
"virtualStoragePartialDownloadsEnabled";
public static final String USE_SHARED_LOCK = "useSharedLock";
public static final String TASK_LOCK_TYPE = "taskLockType";
public static final String USE_CONCURRENT_LOCKS = "useConcurrentLocks";
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/input/DruidInputSource.java
b/indexing-service/src/main/java/org/apache/druid/indexing/input/DruidInputSource.java
index 84fead9916b..917936c03c5 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/input/DruidInputSource.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/input/DruidInputSource.java
@@ -314,7 +314,7 @@ public class DruidInputSource extends AbstractInputSource
implements SplittableI
protected InputSourceReader fixedFormatReader(InputRowSchema inputRowSchema,
@Nullable File temporaryDirectory)
{
final SegmentCacheManager segmentCacheManager =
- segmentCacheManagerFactory.manufacturate(temporaryDirectory, null,
false);
+ segmentCacheManagerFactory.manufacturate(temporaryDirectory, null,
false, false);
final List<TimelineObjectHolder<String, DataSegment>> timeline =
createTimeline();
final Iterator<DruidSegmentInputEntity> entityIterator = FluentIterable
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
index 87622425794..b24f9e6f084 100644
---
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
@@ -19,6 +19,7 @@
package org.apache.druid.indexing.common;
+import com.fasterxml.jackson.databind.InjectableValues;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.inject.Provider;
import com.google.inject.util.Providers;
@@ -27,8 +28,10 @@ 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.SegmentLocalCacheManager;
import org.apache.druid.segment.loading.StorageLoadingThreadPool;
import org.apache.druid.server.metrics.NoopServiceEmitter;
+import org.apache.druid.utils.RuntimeInfo;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
@@ -60,12 +63,17 @@ class SegmentCacheManagerFactoryTest
void testVirtualStorageManagersShareTheInjectedPool()
{
final StorageLoadingThreadPool shared =
- StorageLoadingThreadPool.createFromConfig(new
SegmentLoaderConfig().withVirtualStorage(true));
+
StorageLoadingThreadPool.createFromConfig(SegmentLoaderConfig.builder().build().toEphemeralVirtualStorage());
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);
+ new SegmentCacheManagerFactory(
+ TestIndex.INDEX_IO,
+ jsonMapper,
+ SegmentLoaderConfig.builder().build(),
+ Providers.of(shared)
+ );
+ final SegmentCacheManager m1 = factory.manufacturate(new File(tempDir,
"a"), null, true, false);
+ final SegmentCacheManager m2 = factory.manufacturate(new File(tempDir,
"b"), null, true, false);
Assertions.assertSame(shared, m1.getLoadingThreadPool());
Assertions.assertSame(shared, m2.getLoadingThreadPool());
@@ -84,19 +92,111 @@ class SegmentCacheManagerFactoryTest
throw new AssertionError("ephemeral loading pool must not be resolved
for virtualStorage=false");
};
final SegmentCacheManagerFactory factory =
- new SegmentCacheManagerFactory(TestIndex.INDEX_IO, jsonMapper,
throwingProvider);
+ new SegmentCacheManagerFactory(
+ TestIndex.INDEX_IO,
+ jsonMapper,
+ SegmentLoaderConfig.builder().build(),
+ throwingProvider
+ );
- final SegmentCacheManager m = factory.manufacturate(new File(tempDir,
"c"), null, false);
+ final SegmentCacheManager m = factory.manufacturate(new File(tempDir,
"c"), null, false, false);
Assertions.assertFalse(m.getLoadingThreadPool().isAvailable());
}
+ @Test
+ void testManufacturateDerivesPerTaskConfigFromInjectedConfig() throws
Exception
+ {
+ // The injected druid.segmentCache config carries operator-tuned
virtual-storage settings and (typical of an
+ // indexing node) is not itself in virtual-storage mode. Deriving per-task
caches from it must preserve that tuning.
+ final ObjectMapper mapper = TestHelper.makeJsonMapper();
+ mapper.setInjectableValues(new
InjectableValues.Std().addValue(RuntimeInfo.class, new RuntimeInfo()));
+ final SegmentLoaderConfig injected = mapper.readValue(
+ "{\"virtualStorageMetadataReservationEstimate\": 99999999,"
+ + " \"virtualStorageCoalesceGapBytes\": 65536,"
+ + " \"virtualStorageMaxFetchRunBytes\": 8388608,"
+ + " \"infoDir\": \"/var/druid/segment-cache/info_dir\","
+ + " \"numThreadsToLoadSegmentsIntoPageCacheOnDownload\": 4}",
+ SegmentLoaderConfig.class
+ );
+
+ final StorageLoadingThreadPool shared =
+
StorageLoadingThreadPool.createFromConfig(SegmentLoaderConfig.builder().build().toEphemeralVirtualStorage());
+ try {
+ final SegmentCacheManagerFactory factory =
+ new SegmentCacheManagerFactory(TestIndex.INDEX_IO, jsonMapper,
injected, Providers.of(shared));
+ final SegmentLocalCacheManager m =
+ (SegmentLocalCacheManager) factory.manufacturate(new File(tempDir,
"task"), null, true, true);
+ final SegmentLoaderConfig derived = m.getConfig();
+
+ // Operator-tuned settings are carried over from the injected config...
+ Assertions.assertEquals(99999999L,
derived.getVirtualStorageMetadataReservationEstimate());
+ Assertions.assertEquals(65536L,
derived.getVirtualStorageCoalesceGapBytes());
+ Assertions.assertEquals(8388608L,
derived.getVirtualStorageMaxFetchRunBytes());
+ // ...while the per-task location and mode flags are overridden.
+ Assertions.assertTrue(derived.isVirtualStorage());
+ Assertions.assertTrue(derived.isVirtualStorageEphemeral());
+ Assertions.assertTrue(derived.isVirtualStoragePartialDownloadsEnabled());
+ Assertions.assertEquals(1, derived.getLocations().size());
+ Assertions.assertEquals(new File(tempDir, "task"),
derived.getLocations().get(0).getPath());
+ // ...and settings incompatible with virtual storage are cleared (no
shared info dir, no page-cache warming).
+ Assertions.assertNull(derived.getInfoDir());
+ Assertions.assertEquals(0,
derived.getNumThreadsToLoadSegmentsIntoPageCacheOnDownload());
+
+ // The shared injected config is not mutated by the derive.
+ Assertions.assertFalse(injected.isVirtualStorage());
+ Assertions.assertTrue(injected.getLocations().isEmpty());
+ }
+ finally {
+ shared.stop();
+ }
+ }
+
+ @Test
+ void testNonVirtualManufacturateUsesFreshDefaultsNotInjectedConfig() throws
Exception
+ {
+ // A non-virtual (eager) per-task cache must NOT inherit the node's
transient settings: a positive
+ // numThreadsToLoadSegmentsIntoPageCacheOnDownload would leak a per-reader
executor (input entities drop segments
+ // but never shut the manager down), and deleteOnRemove=false would leave
downloaded segments behind. It starts
+ // from fresh defaults instead.
+ final ObjectMapper mapper = TestHelper.makeJsonMapper();
+ mapper.setInjectableValues(new
InjectableValues.Std().addValue(RuntimeInfo.class, new RuntimeInfo()));
+ final SegmentLoaderConfig injected = mapper.readValue(
+ "{\"numThreadsToLoadSegmentsIntoPageCacheOnDownload\": 4,"
+ + " \"deleteOnRemove\": false,"
+ + " \"lazyLoadOnStart\": true,"
+ + " \"infoDir\": \"/var/druid/segment-cache/info_dir\"}",
+ SegmentLoaderConfig.class
+ );
+
+ // The ephemeral loading pool must not even be resolved for a non-virtual
cache.
+ 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,
injected, throwingProvider);
+
+ final SegmentLocalCacheManager m =
+ (SegmentLocalCacheManager) factory.manufacturate(new File(tempDir,
"eager"), null, false, false);
+ final SegmentLoaderConfig derived = m.getConfig();
+
+ // Transient settings are the safe defaults, not the injected node's
values.
+ Assertions.assertFalse(derived.isVirtualStorage());
+ Assertions.assertEquals(0,
derived.getNumThreadsToLoadSegmentsIntoPageCacheOnDownload());
+ Assertions.assertTrue(derived.isDeleteOnRemove());
+ Assertions.assertNull(derived.getInfoDir());
+ Assertions.assertFalse(derived.isLazyLoadOnStart());
+ // The per-task location is still applied.
+ Assertions.assertEquals(1, derived.getLocations().size());
+ Assertions.assertEquals(new File(tempDir, "eager"),
derived.getLocations().get(0).getPath());
+ }
+
@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);
+ final SegmentCacheManager m1 = factory.manufacturate(new File(tempDir,
"d"), null, true, false);
+ final SegmentCacheManager m2 = factory.manufacturate(new File(tempDir,
"e"), null, true, false);
try {
Assertions.assertTrue(m1.getLoadingThreadPool().isAvailable());
// createWithOwnedPool builds one pool and shares it across the
factory's manufacturate calls.
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/common/TaskToolboxTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/common/TaskToolboxTest.java
index d83ea288586..5933f57bdaf 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/common/TaskToolboxTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/common/TaskToolboxTest.java
@@ -20,6 +20,7 @@
package org.apache.druid.indexing.common;
import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableMap;
import org.apache.druid.client.cache.Cache;
import org.apache.druid.client.cache.CacheConfig;
import org.apache.druid.client.cache.CachePopulatorStats;
@@ -109,6 +110,9 @@ public class TaskToolboxTest
EasyMock.expect(task.getId()).andReturn("task_id").anyTimes();
EasyMock.expect(task.getDataSource()).andReturn("task_ds").anyTimes();
EasyMock.expect(task.getContextValue(Tasks.STORE_EMPTY_COLUMNS_KEY,
true)).andReturn(true).anyTimes();
+ EasyMock.expect(task.getContext())
+
.andReturn(ImmutableMap.of(Tasks.VIRTUAL_STORAGE_PARTIAL_DOWNLOADS_KEY,
"false"))
+ .anyTimes();
IndexMergerV9 indexMergerV9 = EasyMock.createMock(IndexMergerV9.class);
EasyMock.expect(mockIndexMergerV9.create(true)).andReturn(indexMergerV9).anyTimes();
EasyMock.replay(task, mockHandoffNotifierFactory, mockIndexMergerV9);
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/common/config/TaskConfigBuilder.java
b/indexing-service/src/test/java/org/apache/druid/indexing/common/config/TaskConfigBuilder.java
index 69f05e763a0..ecfbdbbd393 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/common/config/TaskConfigBuilder.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/common/config/TaskConfigBuilder.java
@@ -37,6 +37,7 @@ public class TaskConfigBuilder
private boolean enableTaskLevelLogPush;
private Long tmpStorageBytesPerTask;
private boolean buildV10;
+ private Boolean virtualStoragePartialDownloadsEnabled;
public TaskConfigBuilder setBaseDir(String baseDir)
{
@@ -104,6 +105,12 @@ public class TaskConfigBuilder
return this;
}
+ public TaskConfigBuilder setVirtualStoragePartialDownloadsEnabled(Boolean
virtualStoragePartialDownloadsEnabled)
+ {
+ this.virtualStoragePartialDownloadsEnabled =
virtualStoragePartialDownloadsEnabled;
+ return this;
+ }
+
public TaskConfig build()
{
return new TaskConfig(
@@ -117,7 +124,8 @@ public class TaskConfigBuilder
storeEmptyColumns,
enableTaskLevelLogPush,
tmpStorageBytesPerTask,
- buildV10
+ buildV10,
+ virtualStoragePartialDownloadsEnabled
);
}
}
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/common/config/TaskConfigTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/common/config/TaskConfigTest.java
new file mode 100644
index 00000000000..b1a4d0e3b79
--- /dev/null
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/common/config/TaskConfigTest.java
@@ -0,0 +1,45 @@
+/*
+ * 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.config;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.druid.jackson.DefaultObjectMapper;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class TaskConfigTest
+{
+ private final ObjectMapper jsonMapper = new DefaultObjectMapper();
+
+ @Test
+ void testVirtualStoragePartialDownloadsEnabledDefaultsToTrue() throws
Exception
+ {
+ final TaskConfig config = jsonMapper.readValue("{}", TaskConfig.class);
+ Assertions.assertTrue(config.isVirtualStoragePartialDownloadsEnabled());
+ }
+
+ @Test
+ void testVirtualStoragePartialDownloadsEnabledFromJson() throws Exception
+ {
+ final TaskConfig config =
+ jsonMapper.readValue("{\"virtualStoragePartialDownloadsEnabled\":
false}", TaskConfig.class);
+ Assertions.assertFalse(config.isVirtualStoragePartialDownloadsEnabled());
+ }
+}
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 267cc906881..65948d2f405 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
@@ -1268,7 +1268,12 @@ public abstract class CompactionTaskRunBase
Assert.assertEquals(new NumberedShardSpec(0, 1),
segments.get(0).getShardSpec());
final File cacheDir = temporaryFolder.newFolder();
- final SegmentCacheManager segmentCacheManager =
segmentCacheManagerFactory.manufacturate(cacheDir, null, false);
+ final SegmentCacheManager segmentCacheManager =
segmentCacheManagerFactory.manufacturate(
+ cacheDir,
+ null,
+ false,
+ false
+ );
List<String> rowsFromSegment = new ArrayList<>();
for (DataSegment segment : segments) {
@@ -1380,7 +1385,12 @@ public abstract class CompactionTaskRunBase
Assert.assertEquals(new NumberedShardSpec(0, 1),
segments.get(0).getShardSpec());
final File cacheDir = temporaryFolder.newFolder();
- final SegmentCacheManager segmentCacheManager =
segmentCacheManagerFactory.manufacturate(cacheDir, null, false);
+ final SegmentCacheManager segmentCacheManager =
segmentCacheManagerFactory.manufacturate(
+ cacheDir,
+ null,
+ false,
+ false
+ );
List<String> rowsFromSegment = new ArrayList<>();
for (DataSegment segment : segments) {
@@ -1497,7 +1507,12 @@ public abstract class CompactionTaskRunBase
Assert.assertEquals(new NumberedShardSpec(0, 1),
compactSegment.getShardSpec());
final File cacheDir = temporaryFolder.newFolder();
- final SegmentCacheManager segmentCacheManager =
segmentCacheManagerFactory.manufacturate(cacheDir, null, false);
+ final SegmentCacheManager segmentCacheManager =
segmentCacheManagerFactory.manufacturate(
+ cacheDir,
+ null,
+ false,
+ false
+ );
List<String> rowsFromSegment = new ArrayList<>();
segmentCacheManager.load(compactSegment);
@@ -1695,26 +1710,11 @@ public abstract class CompactionTaskRunBase
private TaskToolbox createTaskToolbox(ObjectMapper objectMapper,
TaskActionClient taskActionClient) throws IOException
{
- final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig()
- {
- @Override
- public List<StorageLocationConfig> getLocations()
- {
- return ImmutableList.of(new StorageLocationConfig(localDeepStorage,
null, null));
- }
-
- @Override
- public boolean isVirtualStorage()
- {
- return true;
- }
-
- @Override
- public boolean isVirtualStorageEphemeral()
- {
- return true;
- }
- };
+ final SegmentLoaderConfig loaderConfig = SegmentLoaderConfig.builder()
+ .locations(new StorageLocationConfig(localDeepStorage, null, null))
+ .virtualStorage(true)
+ .virtualStorageIsEphemeral(true)
+ .build();
final List<StorageLocation> storageLocations =
loaderConfig.toStorageLocations();
final SegmentCacheManager cacheManager = new SegmentLocalCacheManager(
storageLocations,
@@ -1754,7 +1754,12 @@ public abstract class CompactionTaskRunBase
protected List<String> getCSVFormatRowsFromSegments(List<DataSegment>
segments) throws Exception
{
final File cacheDir = temporaryFolder.newFolder();
- final SegmentCacheManager segmentCacheManager =
segmentCacheManagerFactory.manufacturate(cacheDir, null, false);
+ final SegmentCacheManager segmentCacheManager =
segmentCacheManagerFactory.manufacturate(
+ cacheDir,
+ null,
+ false,
+ false
+ );
List<String> rowsFromSegment = new ArrayList<>();
for (DataSegment segment : segments) {
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/IndexTaskTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/IndexTaskTest.java
index 399cb9a513e..1aedb848dc5 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/IndexTaskTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/IndexTaskTest.java
@@ -206,16 +206,9 @@ public class IndexTaskTest extends IngestionTestBase
{
final File cacheDir = temporaryFolder.newFolder();
tmpDir = temporaryFolder.newFolder();
- final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig()
- {
- @Override
- public List<StorageLocationConfig> getLocations()
- {
- return Collections.singletonList(
- new StorageLocationConfig(cacheDir, null, null)
- );
- }
- };
+ final SegmentLoaderConfig loaderConfig = SegmentLoaderConfig.builder()
+ .locations(new StorageLocationConfig(cacheDir, null, null))
+ .build();
final List<StorageLocation> storageLocations =
loaderConfig.toStorageLocations();
segmentCacheManager = new SegmentLocalCacheManager(
storageLocations,
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 eb0f1c0b653..aa38dc731db 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
@@ -227,7 +227,7 @@ public abstract class IngestionTestBase extends
InitializedNullHandlingTest
public SegmentCacheManager newSegmentLoader(File storageDir)
{
- return segmentCacheManagerFactory.manufacturate(storageDir, null, true);
+ return segmentCacheManagerFactory.manufacturate(storageDir, null, true,
false);
}
public ObjectMapper getObjectMapper()
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 ef0a730e932..8154da3536c 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
@@ -258,7 +258,7 @@ abstract class AbstractMultiPhaseParallelIndexingTest
extends AbstractParallelIn
private Segment loadSegment(DataSegment dataSegment, File tempSegmentDir)
{
final SegmentCacheManager cacheManager =
SegmentCacheManagerFactory.createWithOwnedPool(TestIndex.INDEX_IO,
getObjectMapper())
- .manufacturate(tempSegmentDir, null, false);
+ .manufacturate(tempSegmentDir, null, false, false);
try {
cacheManager.load(dataSegment);
return cacheManager.acquireCachedSegment(dataSegment.getId(),
AcquireMode.FULL).orElseThrow();
diff --git
a/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/IndexerWorkerContext.java
b/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/IndexerWorkerContext.java
index 3e1c151e5f2..241241bcae0 100644
---
a/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/IndexerWorkerContext.java
+++
b/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/IndexerWorkerContext.java
@@ -174,6 +174,11 @@ public class IndexerWorkerContext implements WorkerContext
{
final IndexIO indexIO = injector.getInstance(IndexIO.class);
final TaskConfig taskConfig = injector.getInstance(TaskConfig.class);
+ // Opt into on-demand partial downloads via the MSQ query context (falls
back to the node's TaskConfig default).
+ final boolean partialDownloadsEnabled =
MultiStageQueryContext.getVirtualStoragePartialDownloadsEnabled(
+ QueryContext.of(task.getContext()),
+ taskConfig.isVirtualStoragePartialDownloadsEnabled()
+ );
final SegmentCacheManager cacheManager =
injector.getInstance(SegmentCacheManagerFactory.class)
.manufacturate(
@@ -181,7 +186,8 @@ public class IndexerWorkerContext implements WorkerContext
// Divide tmpStorageBytesPerTask by 3 so the local cache
never takes up the majority of space.
// In a typical leaf stage run, we may need some disk
space for inputs and some for outputs.
taskConfig.getTmpStorageBytesPerTask() > 0 ?
taskConfig.getTmpStorageBytesPerTask() / 3 : null,
- true
+ true,
+ partialDownloadsEnabled
);
final SegmentManager segmentManager = new SegmentManager(cacheManager);
final VirtualStorageManager virtualStorageManager =
diff --git
a/multi-stage-query/src/main/java/org/apache/druid/msq/util/MultiStageQueryContext.java
b/multi-stage-query/src/main/java/org/apache/druid/msq/util/MultiStageQueryContext.java
index 114e70a7132..cf840e7bc03 100644
---
a/multi-stage-query/src/main/java/org/apache/druid/msq/util/MultiStageQueryContext.java
+++
b/multi-stage-query/src/main/java/org/apache/druid/msq/util/MultiStageQueryContext.java
@@ -235,6 +235,13 @@ public class MultiStageQueryContext
public static final String CTX_INCLUDE_ALL_COUNTERS = "includeAllCounters";
public static final boolean DEFAULT_INCLUDE_ALL_COUNTERS = true;
+ /**
+ * Whether worker tasks read input segments via on-demand partial
(per-column) downloads instead of downloading each
+ * segment in full. Aliases the task-context key {@link
Tasks#VIRTUAL_STORAGE_PARTIAL_DOWNLOADS_KEY} so it can be set
+ * on the MSQ query context; when unset it falls back to the node's {@code
TaskConfig} default.
+ */
+ public static final String CTX_VIRTUAL_STORAGE_PARTIAL_DOWNLOADS =
Tasks.VIRTUAL_STORAGE_PARTIAL_DOWNLOADS_KEY;
+
/**
* Whether workers should send live counter updates to the controller via
the message relay. When enabled, workers
* periodically send counter snapshots to the controller, allowing the
controller to have more up-to-date progress
@@ -322,6 +329,14 @@ public class MultiStageQueryContext
);
}
+ public static boolean getVirtualStoragePartialDownloadsEnabled(
+ final QueryContext queryContext,
+ final boolean defaultValue
+ )
+ {
+ return queryContext.getBoolean(CTX_VIRTUAL_STORAGE_PARTIAL_DOWNLOADS,
defaultValue);
+ }
+
public static boolean isDurableStorageEnabled(final QueryContext
queryContext)
{
return queryContext.getBoolean(
diff --git
a/multi-stage-query/src/test/java/org/apache/druid/msq/input/RegularLoadableSegmentTest.java
b/multi-stage-query/src/test/java/org/apache/druid/msq/input/RegularLoadableSegmentTest.java
index 003d8a6275c..9fa2630f424 100644
---
a/multi-stage-query/src/test/java/org/apache/druid/msq/input/RegularLoadableSegmentTest.java
+++
b/multi-stage-query/src/test/java/org/apache/druid/msq/input/RegularLoadableSegmentTest.java
@@ -24,7 +24,6 @@ import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.annotation.JsonTypeName;
import com.fasterxml.jackson.databind.InjectableValues;
import com.fasterxml.jackson.databind.ObjectMapper;
-import com.google.common.collect.ImmutableList;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
@@ -164,11 +163,12 @@ class RegularLoadableSegmentTest extends
InitializedNullHandlingTest
// SegmentManager with virtualStorage for dynamically-loaded data tests
cacheDir = tempDir.resolve("cache").toFile();
- final SegmentLoaderConfig virtualLoaderConfig = new SegmentLoaderConfig()
- .setLocations(ImmutableList.of(new StorageLocationConfig(cacheDir,
10_000_000_000L, null)))
- .setVirtualStorage(true)
- .setVirtualStorageIsEphemeral(true)
- .setVirtualStoragePartialDownloadsEnabled(true);
+ final SegmentLoaderConfig virtualLoaderConfig =
SegmentLoaderConfig.builder()
+ .locations(new StorageLocationConfig(cacheDir, 10_000_000_000L, null))
+ .virtualStorage(true)
+ .virtualStorageIsEphemeral(true)
+ .virtualStoragePartialDownloadsEnabled(true)
+ .build();
final List<StorageLocation> virtualLocations =
virtualLoaderConfig.toStorageLocations();
segmentManagerDynamic = new SegmentManager(
new SegmentLocalCacheManager(
@@ -183,8 +183,9 @@ class RegularLoadableSegmentTest extends
InitializedNullHandlingTest
// SegmentManager without virtualStorage for pre-loaded data tests
preLoadCacheDir = tempDir.resolve("localCache").toFile();
- final SegmentLoaderConfig localLoaderConfig = new SegmentLoaderConfig()
- .setLocations(ImmutableList.of(new
StorageLocationConfig(preLoadCacheDir, 10_000_000_000L, null)));
+ final SegmentLoaderConfig localLoaderConfig = SegmentLoaderConfig.builder()
+ .locations(new StorageLocationConfig(preLoadCacheDir, 10_000_000_000L,
null))
+ .build();
final List<StorageLocation> localLocations =
localLoaderConfig.toStorageLocations();
segmentManagerPreLoad = new SegmentManager(
new SegmentLocalCacheManager(
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 15d74d025c4..5827a71c50f 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
@@ -95,7 +95,7 @@ public class CalciteMSQTestsHelper
public SegmentCacheManager provideSegmentCacheManager(ObjectMapper
testMapper, TempDirProducer tempDirProducer)
{
return
SegmentCacheManagerFactory.createWithOwnedPool(TestIndex.INDEX_IO, testMapper)
- .manufacturate(tempDirProducer.newTempFolder("test"), null, true);
+ .manufacturate(tempDirProducer.newTempFolder("test"), null, true,
false);
}
@Provides
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 19ae4829f59..f0381b25c23 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
@@ -474,7 +474,8 @@ public class MSQTestBase extends BaseCalciteQueryTest
indexIO = new IndexIO(objectMapper, ColumnConfig.DEFAULT);
segmentCacheManager =
- SegmentCacheManagerFactory.createWithOwnedPool(indexIO,
objectMapper).manufacturate(newTempFolder("cacheManager"), null, true);
+ SegmentCacheManagerFactory.createWithOwnedPool(indexIO, objectMapper)
+
.manufacturate(newTempFolder("cacheManager"), null, true, false);
testSegmentManager = new TestSegmentManager();
diff --git
a/multi-stage-query/src/test/java/org/apache/druid/msq/util/MultiStageQueryContextTest.java
b/multi-stage-query/src/test/java/org/apache/druid/msq/util/MultiStageQueryContextTest.java
index 6db3597d632..44e05c0a81a 100644
---
a/multi-stage-query/src/test/java/org/apache/druid/msq/util/MultiStageQueryContextTest.java
+++
b/multi-stage-query/src/test/java/org/apache/druid/msq/util/MultiStageQueryContextTest.java
@@ -470,6 +470,30 @@ public class MultiStageQueryContextTest
Assert.assertEquals(Integer.valueOf(4),
MultiStageQueryContext.getMaxThreads(QueryContext.of(propertyMap)));
}
+ @Test
+ public void getVirtualStoragePartialDownloadsEnabled_unset_returnsDefault()
+ {
+
Assert.assertFalse(MultiStageQueryContext.getVirtualStoragePartialDownloadsEnabled(QueryContext.empty(),
false));
+
Assert.assertTrue(MultiStageQueryContext.getVirtualStoragePartialDownloadsEnabled(QueryContext.empty(),
true));
+ }
+
+ @Test
+ public void getVirtualStoragePartialDownloadsEnabled_set_overridesDefault()
+ {
+ Assert.assertTrue(
+ MultiStageQueryContext.getVirtualStoragePartialDownloadsEnabled(
+
QueryContext.of(ImmutableMap.of(MultiStageQueryContext.CTX_VIRTUAL_STORAGE_PARTIAL_DOWNLOADS,
true)),
+ false
+ )
+ );
+ Assert.assertFalse(
+ MultiStageQueryContext.getVirtualStoragePartialDownloadsEnabled(
+
QueryContext.of(ImmutableMap.of(MultiStageQueryContext.CTX_VIRTUAL_STORAGE_PARTIAL_DOWNLOADS,
false)),
+ true
+ )
+ );
+ }
+
@Test
public void withCommonContext_noTimeout_setsStartTimeOnly()
{
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 24d4e360c1e..8feeb2cc25b 100644
--- a/server/src/main/java/org/apache/druid/guice/StorageNodeModule.java
+++ b/server/src/main/java/org/apache/druid/guice/StorageNodeModule.java
@@ -146,7 +146,7 @@ public class StorageNodeModule implements Module
{
// 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));
+ return
StorageLoadingThreadPool.createFromConfig(config.toEphemeralVirtualStorage());
}
@Provides
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 b9c9f0804fe..75d1264ea05 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
@@ -20,63 +20,74 @@
package org.apache.druid.segment.loading;
import com.fasterxml.jackson.annotation.JacksonInject;
+import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
-import com.google.common.collect.Lists;
+import org.apache.druid.common.config.Configs;
import org.apache.druid.segment.file.PartialSegmentFileMapperV10;
import org.apache.druid.utils.RuntimeInfo;
+import javax.annotation.Nullable;
import java.io.File;
+import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
/**
- *
+ * Configuration for a local segment cache, bound from {@code
druid.segmentCache}.
*/
public class SegmentLoaderConfig
{
- @JacksonInject
- private final RuntimeInfo runtimeInfo = new RuntimeInfo();
-
- @JsonProperty
- private List<StorageLocationConfig> locations = Collections.emptyList();
-
+ private static final boolean DEFAULT_LAZY_LOAD_ON_START = false;
+ private static final boolean DEFAULT_DELETE_ON_REMOVE = true;
+ private static final int DEFAULT_DROP_SEGMENT_DELAY_MILLIS = (int)
TimeUnit.SECONDS.toMillis(30);
+ private static final int DEFAULT_ANNOUNCE_INTERVAL_MILLIS = 0; // do not
background announce
+ private static final int
DEFAULT_NUM_THREADS_TO_LOAD_INTO_PAGE_CACHE_ON_DOWNLOAD = 0;
+ private static final int DEFAULT_STATUS_QUEUE_MAX_SIZE = 100;
+ private static final boolean DEFAULT_VIRTUAL_STORAGE = false;
+ private static final boolean DEFAULT_VIRTUAL_STORAGE_USE_VIRTUAL_THREADS =
true;
+ private static final boolean DEFAULT_VIRTUAL_STORAGE_IS_EPHEMERAL = false;
+ private static final long
DEFAULT_VIRTUAL_STORAGE_METADATA_RESERVATION_ESTIMATE = 16L * 1024L * 1024L;
+ private static final boolean
DEFAULT_VIRTUAL_STORAGE_PARTIAL_DOWNLOADS_ENABLED = false;
+
+ private final RuntimeInfo runtimeInfo;
+
+ @JsonProperty("locations")
+ private final List<StorageLocationConfig> locations;
@JsonProperty("lazyLoadOnStart")
- private boolean lazyLoadOnStart = false;
-
+ private final boolean lazyLoadOnStart;
@JsonProperty("deleteOnRemove")
- private boolean deleteOnRemove = true;
-
+ private final boolean deleteOnRemove;
@JsonProperty("dropSegmentDelayMillis")
- private int dropSegmentDelayMillis = (int) TimeUnit.SECONDS.toMillis(30);
-
+ private final int dropSegmentDelayMillis;
@JsonProperty("announceIntervalMillis")
- private int announceIntervalMillis = 0; // do not background announce
-
+ private final int announceIntervalMillis;
@JsonProperty("numLoadingThreads")
- private int numLoadingThreads = Math.max(1,
runtimeInfo.getAvailableProcessors() / 6);
-
+ private final int numLoadingThreads;
+ /**
+ * Nullable so {@link #getNumBootstrapThreads()} can fall back to {@link
#numLoadingThreads} when unset.
+ */
@JsonProperty("numBootstrapThreads")
- private Integer numBootstrapThreads = null;
-
+ @Nullable
+ private final Integer numBootstrapThreads;
@JsonProperty("numThreadsToLoadSegmentsIntoPageCacheOnDownload")
- private int numThreadsToLoadSegmentsIntoPageCacheOnDownload = 0;
-
+ private final int numThreadsToLoadSegmentsIntoPageCacheOnDownload;
+ /**
+ * Nullable so {@link
#getNumThreadsToLoadSegmentsIntoPageCacheOnBootstrap()} can fall back to the
on-download value.
+ */
@JsonProperty("numThreadsToLoadSegmentsIntoPageCacheOnBootstrap")
- private Integer numThreadsToLoadSegmentsIntoPageCacheOnBootstrap = null;
-
- @JsonProperty
- private File infoDir = null;
-
- @JsonProperty
- private int statusQueueMaxSize = 100;
-
+ @Nullable
+ private final Integer numThreadsToLoadSegmentsIntoPageCacheOnBootstrap;
+ @JsonProperty("infoDir")
+ @Nullable
+ private final File infoDir;
+ @JsonProperty("statusQueueMaxSize")
+ private final int statusQueueMaxSize;
@JsonProperty("virtualStorage")
- private boolean virtualStorage = false;
-
+ private final boolean virtualStorage;
@JsonProperty("virtualStorageLoadThreads")
- private int virtualStorageLoadThreads = Math.max(32, 4 *
runtimeInfo.getAvailableProcessors());
+ private final int virtualStorageLoadThreads;
/**
* When true (the default), the on-demand load executor uses one virtual
thread per task with a {@link
@@ -85,15 +96,15 @@ public class SegmentLoaderConfig
* particular deep storage SDK or workload.
*/
@JsonProperty("virtualStorageUseVirtualThreads")
- private boolean virtualStorageUseVirtualThreads = true;
+ private final boolean virtualStorageUseVirtualThreads;
/**
* When enabled, weakly-held cache entries are evicted immediately upon
release of all holds, rather than
* waiting for space pressure to trigger eviction. This setting is not
intended to be configured directly by
- * administrators. Instead, it is expected to be set when appropriate via
{@link #setVirtualStorage}.
+ * administrators. Instead, it is expected to be set when appropriate via
{@link Builder#virtualStorageIsEphemeral}.
*/
@JsonProperty("virtualStorageIsEphemeral")
- private boolean virtualStorageIsEphemeral = false;
+ private final boolean virtualStorageIsEphemeral;
/**
* Up-front size reservation (in bytes) used when mounting a partial-segment
metadata cache entry. The entry
@@ -104,7 +115,7 @@ public class SegmentLoaderConfig
* need a higher value.
*/
@JsonProperty("virtualStorageMetadataReservationEstimate")
- private long virtualStorageMetadataReservationEstimate = 16L * 1024L * 1024L;
+ private final long virtualStorageMetadataReservationEstimate;
/**
* When true, partial-eligible V10 segments are mounted via the partial
machinery and
@@ -113,7 +124,7 @@ public class SegmentLoaderConfig
* {@link AcquireMode#FULL} so the entire segment is downloaded up front
(matching pre-partial-download behavior).
*/
@JsonProperty("virtualStoragePartialDownloadsEnabled")
- private boolean virtualStoragePartialDownloadsEnabled = false;
+ private final boolean virtualStoragePartialDownloadsEnabled;
/**
* Maximum number of unrequested bytes a partial-download range read will
fetch in order to bridge two requested
@@ -124,7 +135,7 @@ public class SegmentLoaderConfig
* request latency vs streaming throughput.
*/
@JsonProperty("virtualStorageCoalesceGapBytes")
- private long virtualStorageCoalesceGapBytes =
PartialSegmentFileMapperV10.DEFAULT_COALESCE_GAP_BYTES;
+ private final long virtualStorageCoalesceGapBytes;
/**
* Maximum size of a single range read in a query-driven partial download.
Larger fetches split at internal-file
@@ -133,36 +144,82 @@ public class SegmentLoaderConfig
* containers sequentially and are unaffected. {@code <= 0} disables
splitting. Defaults to 64 MiB.
*/
@JsonProperty("virtualStorageMaxFetchRunBytes")
- private long virtualStorageMaxFetchRunBytes =
PartialSegmentFileMapperV10.DEFAULT_MAX_FETCH_RUN_BYTES;
-
- private long combinedMaxSize = 0;
-
- public SegmentLoaderConfig()
- {
- }
-
- private SegmentLoaderConfig(SegmentLoaderConfig other)
+ private final long virtualStorageMaxFetchRunBytes;
+
+ @JsonCreator
+ public SegmentLoaderConfig(
+ @JacksonInject @Nullable RuntimeInfo runtimeInfo,
+ @JsonProperty("locations") @Nullable List<StorageLocationConfig>
locations,
+ @JsonProperty("lazyLoadOnStart") @Nullable Boolean lazyLoadOnStart,
+ @JsonProperty("deleteOnRemove") @Nullable Boolean deleteOnRemove,
+ @JsonProperty("dropSegmentDelayMillis") @Nullable Integer
dropSegmentDelayMillis,
+ @JsonProperty("announceIntervalMillis") @Nullable Integer
announceIntervalMillis,
+ @JsonProperty("numLoadingThreads") @Nullable Integer numLoadingThreads,
+ @JsonProperty("numBootstrapThreads") @Nullable Integer
numBootstrapThreads,
+ @JsonProperty("numThreadsToLoadSegmentsIntoPageCacheOnDownload")
@Nullable Integer numThreadsToLoadSegmentsIntoPageCacheOnDownload,
+ @JsonProperty("numThreadsToLoadSegmentsIntoPageCacheOnBootstrap")
@Nullable Integer numThreadsToLoadSegmentsIntoPageCacheOnBootstrap,
+ @JsonProperty("infoDir") @Nullable File infoDir,
+ @JsonProperty("statusQueueMaxSize") @Nullable Integer statusQueueMaxSize,
+ @JsonProperty("virtualStorage") @Nullable Boolean virtualStorage,
+ @JsonProperty("virtualStorageLoadThreads") @Nullable Integer
virtualStorageLoadThreads,
+ @JsonProperty("virtualStorageUseVirtualThreads") @Nullable Boolean
virtualStorageUseVirtualThreads,
+ @JsonProperty("virtualStorageIsEphemeral") @Nullable Boolean
virtualStorageIsEphemeral,
+ @JsonProperty("virtualStorageMetadataReservationEstimate") @Nullable
Long virtualStorageMetadataReservationEstimate,
+ @JsonProperty("virtualStoragePartialDownloadsEnabled") @Nullable Boolean
virtualStoragePartialDownloadsEnabled,
+ @JsonProperty("virtualStorageCoalesceGapBytes") @Nullable Long
virtualStorageCoalesceGapBytes,
+ @JsonProperty("virtualStorageMaxFetchRunBytes") @Nullable Long
virtualStorageMaxFetchRunBytes
+ )
{
- 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;
+ // RuntimeInfo is an injected @LazySingleton (test-overridable); fall back
to a fresh instance when it is not
+ // available, e.g. when deserializing outside a Guice context. Only used
to size the thread-count defaults.
+ final RuntimeInfo resolvedRuntimeInfo = runtimeInfo == null ? new
RuntimeInfo() : runtimeInfo;
+ this.runtimeInfo = resolvedRuntimeInfo;
+ this.locations = locations == null ? Collections.emptyList() :
List.copyOf(locations);
+ this.lazyLoadOnStart = Configs.valueOrDefault(lazyLoadOnStart,
DEFAULT_LAZY_LOAD_ON_START);
+ this.deleteOnRemove = Configs.valueOrDefault(deleteOnRemove,
DEFAULT_DELETE_ON_REMOVE);
+ this.dropSegmentDelayMillis =
Configs.valueOrDefault(dropSegmentDelayMillis,
DEFAULT_DROP_SEGMENT_DELAY_MILLIS);
+ this.announceIntervalMillis =
Configs.valueOrDefault(announceIntervalMillis,
DEFAULT_ANNOUNCE_INTERVAL_MILLIS);
+ this.numLoadingThreads = Configs.valueOrDefault(
+ numLoadingThreads,
+ Math.max(1, resolvedRuntimeInfo.getAvailableProcessors() / 6)
+ );
+ this.numBootstrapThreads = numBootstrapThreads;
+ this.numThreadsToLoadSegmentsIntoPageCacheOnDownload =
Configs.valueOrDefault(
+ numThreadsToLoadSegmentsIntoPageCacheOnDownload,
+ DEFAULT_NUM_THREADS_TO_LOAD_INTO_PAGE_CACHE_ON_DOWNLOAD
+ );
+ this.numThreadsToLoadSegmentsIntoPageCacheOnBootstrap =
numThreadsToLoadSegmentsIntoPageCacheOnBootstrap;
+ this.infoDir = infoDir;
+ this.statusQueueMaxSize = Configs.valueOrDefault(statusQueueMaxSize,
DEFAULT_STATUS_QUEUE_MAX_SIZE);
+ this.virtualStorage = Configs.valueOrDefault(virtualStorage,
DEFAULT_VIRTUAL_STORAGE);
+ this.virtualStorageLoadThreads = Configs.valueOrDefault(
+ virtualStorageLoadThreads,
+ Math.max(32, 4 * resolvedRuntimeInfo.getAvailableProcessors())
+ );
+ this.virtualStorageUseVirtualThreads = Configs.valueOrDefault(
+ virtualStorageUseVirtualThreads,
+ DEFAULT_VIRTUAL_STORAGE_USE_VIRTUAL_THREADS
+ );
+ this.virtualStorageIsEphemeral = Configs.valueOrDefault(
+ virtualStorageIsEphemeral,
+ DEFAULT_VIRTUAL_STORAGE_IS_EPHEMERAL
+ );
+ this.virtualStorageMetadataReservationEstimate = Configs.valueOrDefault(
+ virtualStorageMetadataReservationEstimate,
+ DEFAULT_VIRTUAL_STORAGE_METADATA_RESERVATION_ESTIMATE
+ );
+ this.virtualStoragePartialDownloadsEnabled = Configs.valueOrDefault(
+ virtualStoragePartialDownloadsEnabled,
+ DEFAULT_VIRTUAL_STORAGE_PARTIAL_DOWNLOADS_ENABLED
+ );
+ this.virtualStorageCoalesceGapBytes = Configs.valueOrDefault(
+ virtualStorageCoalesceGapBytes,
+ PartialSegmentFileMapperV10.DEFAULT_COALESCE_GAP_BYTES
+ );
+ this.virtualStorageMaxFetchRunBytes = Configs.valueOrDefault(
+ virtualStorageMaxFetchRunBytes,
+ PartialSegmentFileMapperV10.DEFAULT_MAX_FETCH_RUN_BYTES
+ );
}
public List<StorageLocationConfig> getLocations()
@@ -212,6 +269,7 @@ public class SegmentLoaderConfig
numThreadsToLoadSegmentsIntoPageCacheOnBootstrap;
}
+ @Nullable
public File getInfoDir()
{
return infoDir;
@@ -224,10 +282,7 @@ public class SegmentLoaderConfig
public long getCombinedMaxSize()
{
- if (combinedMaxSize == 0) {
- combinedMaxSize =
getLocations().stream().mapToLong(StorageLocationConfig::getMaxSize).sum();
- }
- return combinedMaxSize;
+ return
getLocations().stream().mapToLong(StorageLocationConfig::getMaxSize).sum();
}
public boolean isVirtualStorage()
@@ -270,53 +325,29 @@ public class SegmentLoaderConfig
return virtualStorageMaxFetchRunBytes;
}
- public SegmentLoaderConfig setLocations(List<StorageLocationConfig>
locations)
- {
- this.locations = Lists.newArrayList(locations);
- return this;
- }
-
- public SegmentLoaderConfig setVirtualStoragePartialDownloadsEnabled(boolean
enabled)
- {
- this.virtualStoragePartialDownloadsEnabled = enabled;
- return this;
- }
-
/**
- * Sets {@link #virtualStorage}.
+ * Returns a copy of this config configured as an ephemeral, on-demand
virtual-storage cache:
+ * {@link #isVirtualStorage()} and {@link #isVirtualStorageEphemeral()} are
set, and the settings that only apply to
+ * a classic on-disk historical cache are dropped. Virtual-storage tuning
({@link #getVirtualStorageLoadThreads()},
+ * {@link #isVirtualStorageUseVirtualThreads()}, {@link
#getVirtualStorageMetadataReservationEstimate()}, the
+ * coalescing/fetch-run limits) is preserved, so a per-task cache derived
from a node's {@code druid.segmentCache}
+ * config keeps operator tuning.
*/
- public SegmentLoaderConfig setVirtualStorage(boolean virtualStorage)
+ public SegmentLoaderConfig toEphemeralVirtualStorage()
{
- this.virtualStorage = virtualStorage;
- 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}.
- */
- public SegmentLoaderConfig setVirtualStorageIsEphemeral(boolean
virtualStorageFabricEphemeral)
- {
- this.virtualStorageIsEphemeral = virtualStorageFabricEphemeral;
- return this;
+ return toBuilder()
+ .virtualStorage(true)
+ .virtualStorageIsEphemeral(true)
+ .lazyLoadOnStart(false)
+ .infoDir(null)
+ .numThreadsToLoadSegmentsIntoPageCacheOnDownload(0)
+ .numThreadsToLoadSegmentsIntoPageCacheOnBootstrap(null)
+ .deleteOnRemove(true)
+ .build();
}
/**
* Convert a list of {@link StorageLocationConfig} objects to {@link
StorageLocation} objects.
- * <p>
- * Note: {@link #getLocations} is called instead of variable access because
some testcases overrides this method
*/
public List<StorageLocation> toStorageLocations()
{
@@ -338,6 +369,31 @@ public class SegmentLoaderConfig
.collect(Collectors.toList());
}
+ public Builder toBuilder()
+ {
+ return new Builder()
+ .runtimeInfo(runtimeInfo)
+ .locations(locations)
+ .lazyLoadOnStart(lazyLoadOnStart)
+ .deleteOnRemove(deleteOnRemove)
+ .dropSegmentDelayMillis(dropSegmentDelayMillis)
+ .announceIntervalMillis(announceIntervalMillis)
+ .numLoadingThreads(numLoadingThreads)
+ .numBootstrapThreads(numBootstrapThreads)
+
.numThreadsToLoadSegmentsIntoPageCacheOnDownload(numThreadsToLoadSegmentsIntoPageCacheOnDownload)
+
.numThreadsToLoadSegmentsIntoPageCacheOnBootstrap(numThreadsToLoadSegmentsIntoPageCacheOnBootstrap)
+ .infoDir(infoDir)
+ .statusQueueMaxSize(statusQueueMaxSize)
+ .virtualStorage(virtualStorage)
+ .virtualStorageLoadThreads(virtualStorageLoadThreads)
+ .virtualStorageUseVirtualThreads(virtualStorageUseVirtualThreads)
+ .virtualStorageIsEphemeral(virtualStorageIsEphemeral)
+
.virtualStorageMetadataReservationEstimate(virtualStorageMetadataReservationEstimate)
+
.virtualStoragePartialDownloadsEnabled(virtualStoragePartialDownloadsEnabled)
+ .virtualStorageCoalesceGapBytes(virtualStorageCoalesceGapBytes)
+ .virtualStorageMaxFetchRunBytes(virtualStorageMaxFetchRunBytes);
+ }
+
@Override
public String toString()
{
@@ -361,7 +417,212 @@ public class SegmentLoaderConfig
", virtualStoragePartialDownloadsEnabled=" +
virtualStoragePartialDownloadsEnabled +
", virtualStorageCoalesceGapBytes=" +
virtualStorageCoalesceGapBytes +
", virtualStorageMaxFetchRunBytes=" +
virtualStorageMaxFetchRunBytes +
- ", combinedMaxSize=" + combinedMaxSize +
'}';
}
+
+ public static Builder builder()
+ {
+ return new Builder();
+ }
+
+ /**
+ * Builds a {@link SegmentLoaderConfig}. Any field left unset takes the same
default the {@link JsonCreator}
+ * constructor applies. Obtain one via {@link SegmentLoaderConfig#builder()}
(all defaults) or
+ * {@link SegmentLoaderConfig#toBuilder()} (a copy of an existing config).
+ */
+ public static class Builder
+ {
+ @Nullable
+ private RuntimeInfo runtimeInfo;
+ @Nullable
+ private List<StorageLocationConfig> locations;
+ @Nullable
+ private Boolean lazyLoadOnStart;
+ @Nullable
+ private Boolean deleteOnRemove;
+ @Nullable
+ private Integer dropSegmentDelayMillis;
+ @Nullable
+ private Integer announceIntervalMillis;
+ @Nullable
+ private Integer numLoadingThreads;
+ @Nullable
+ private Integer numBootstrapThreads;
+ @Nullable
+ private Integer numThreadsToLoadSegmentsIntoPageCacheOnDownload;
+ @Nullable
+ private Integer numThreadsToLoadSegmentsIntoPageCacheOnBootstrap;
+ @Nullable
+ private File infoDir;
+ @Nullable
+ private Integer statusQueueMaxSize;
+ @Nullable
+ private Boolean virtualStorage;
+ @Nullable
+ private Integer virtualStorageLoadThreads;
+ @Nullable
+ private Boolean virtualStorageUseVirtualThreads;
+ @Nullable
+ private Boolean virtualStorageIsEphemeral;
+ @Nullable
+ private Long virtualStorageMetadataReservationEstimate;
+ @Nullable
+ private Boolean virtualStoragePartialDownloadsEnabled;
+ @Nullable
+ private Long virtualStorageCoalesceGapBytes;
+ @Nullable
+ private Long virtualStorageMaxFetchRunBytes;
+
+ public Builder runtimeInfo(@Nullable RuntimeInfo runtimeInfo)
+ {
+ this.runtimeInfo = runtimeInfo;
+ return this;
+ }
+
+ public Builder locations(@Nullable List<StorageLocationConfig> locations)
+ {
+ this.locations = locations;
+ return this;
+ }
+
+ public Builder locations(StorageLocationConfig... locations)
+ {
+ this.locations = Arrays.asList(locations);
+ return this;
+ }
+
+ public Builder lazyLoadOnStart(boolean lazyLoadOnStart)
+ {
+ this.lazyLoadOnStart = lazyLoadOnStart;
+ return this;
+ }
+
+ public Builder deleteOnRemove(boolean deleteOnRemove)
+ {
+ this.deleteOnRemove = deleteOnRemove;
+ return this;
+ }
+
+ public Builder dropSegmentDelayMillis(int dropSegmentDelayMillis)
+ {
+ this.dropSegmentDelayMillis = dropSegmentDelayMillis;
+ return this;
+ }
+
+ public Builder announceIntervalMillis(int announceIntervalMillis)
+ {
+ this.announceIntervalMillis = announceIntervalMillis;
+ return this;
+ }
+
+ public Builder numLoadingThreads(int numLoadingThreads)
+ {
+ this.numLoadingThreads = numLoadingThreads;
+ return this;
+ }
+
+ public Builder numBootstrapThreads(@Nullable Integer numBootstrapThreads)
+ {
+ this.numBootstrapThreads = numBootstrapThreads;
+ return this;
+ }
+
+ public Builder numThreadsToLoadSegmentsIntoPageCacheOnDownload(int
numThreads)
+ {
+ this.numThreadsToLoadSegmentsIntoPageCacheOnDownload = numThreads;
+ return this;
+ }
+
+ public Builder numThreadsToLoadSegmentsIntoPageCacheOnBootstrap(@Nullable
Integer numThreads)
+ {
+ this.numThreadsToLoadSegmentsIntoPageCacheOnBootstrap = numThreads;
+ return this;
+ }
+
+ public Builder infoDir(@Nullable File infoDir)
+ {
+ this.infoDir = infoDir;
+ return this;
+ }
+
+ public Builder statusQueueMaxSize(int statusQueueMaxSize)
+ {
+ this.statusQueueMaxSize = statusQueueMaxSize;
+ return this;
+ }
+
+ public Builder virtualStorage(boolean virtualStorage)
+ {
+ this.virtualStorage = virtualStorage;
+ return this;
+ }
+
+ public Builder virtualStorageLoadThreads(int virtualStorageLoadThreads)
+ {
+ this.virtualStorageLoadThreads = virtualStorageLoadThreads;
+ return this;
+ }
+
+ public Builder virtualStorageUseVirtualThreads(boolean
virtualStorageUseVirtualThreads)
+ {
+ this.virtualStorageUseVirtualThreads = virtualStorageUseVirtualThreads;
+ return this;
+ }
+
+ public Builder virtualStorageIsEphemeral(boolean virtualStorageIsEphemeral)
+ {
+ this.virtualStorageIsEphemeral = virtualStorageIsEphemeral;
+ return this;
+ }
+
+ public Builder virtualStorageMetadataReservationEstimate(long
virtualStorageMetadataReservationEstimate)
+ {
+ this.virtualStorageMetadataReservationEstimate =
virtualStorageMetadataReservationEstimate;
+ return this;
+ }
+
+ public Builder virtualStoragePartialDownloadsEnabled(boolean
virtualStoragePartialDownloadsEnabled)
+ {
+ this.virtualStoragePartialDownloadsEnabled =
virtualStoragePartialDownloadsEnabled;
+ return this;
+ }
+
+ public Builder virtualStorageCoalesceGapBytes(long
virtualStorageCoalesceGapBytes)
+ {
+ this.virtualStorageCoalesceGapBytes = virtualStorageCoalesceGapBytes;
+ return this;
+ }
+
+ public Builder virtualStorageMaxFetchRunBytes(long
virtualStorageMaxFetchRunBytes)
+ {
+ this.virtualStorageMaxFetchRunBytes = virtualStorageMaxFetchRunBytes;
+ return this;
+ }
+
+ public SegmentLoaderConfig build()
+ {
+ return new SegmentLoaderConfig(
+ runtimeInfo,
+ locations,
+ lazyLoadOnStart,
+ deleteOnRemove,
+ dropSegmentDelayMillis,
+ announceIntervalMillis,
+ numLoadingThreads,
+ numBootstrapThreads,
+ numThreadsToLoadSegmentsIntoPageCacheOnDownload,
+ numThreadsToLoadSegmentsIntoPageCacheOnBootstrap,
+ infoDir,
+ statusQueueMaxSize,
+ virtualStorage,
+ virtualStorageLoadThreads,
+ virtualStorageUseVirtualThreads,
+ virtualStorageIsEphemeral,
+ virtualStorageMetadataReservationEstimate,
+ virtualStoragePartialDownloadsEnabled,
+ virtualStorageCoalesceGapBytes,
+ virtualStorageMaxFetchRunBytes
+ );
+ }
+ }
}
diff --git
a/server/src/main/java/org/apache/druid/segment/loading/SegmentLocalCacheManager.java
b/server/src/main/java/org/apache/druid/segment/loading/SegmentLocalCacheManager.java
index b48e4202f9b..8ee13001f23 100644
---
a/server/src/main/java/org/apache/druid/segment/loading/SegmentLocalCacheManager.java
+++
b/server/src/main/java/org/apache/druid/segment/loading/SegmentLocalCacheManager.java
@@ -1595,6 +1595,12 @@ public class SegmentLocalCacheManager implements
SegmentCacheManager
return virtualStorageLoadingThreadPool;
}
+ @VisibleForTesting
+ public SegmentLoaderConfig getConfig()
+ {
+ return config;
+ }
+
/**
* Checks whether a segment is already cached. This method does not confirm
if the segment is actually mounted in
* the location, or even that the segment files in some location are valid,
just that some files exist in the
diff --git
a/server/src/test/java/org/apache/druid/client/DruidServerConfigTest.java
b/server/src/test/java/org/apache/druid/client/DruidServerConfigTest.java
index 185685daab8..258fec51b9c 100644
--- a/server/src/test/java/org/apache/druid/client/DruidServerConfigTest.java
+++ b/server/src/test/java/org/apache/druid/client/DruidServerConfigTest.java
@@ -87,7 +87,7 @@ public class DruidServerConfigTest
locations.add(locationConfig2);
DruidServerConfig druidServerConfig = new DruidServerConfig(
new RuntimeInfo(),
- new SegmentLoaderConfig().setLocations(locations)
+ SegmentLoaderConfig.builder().locations(locations).build()
);
Assert.assertEquals(30000000000L, druidServerConfig.getMaxSize());
}
@@ -103,14 +103,7 @@ public class DruidServerConfigTest
return HumanReadableBytes.parse("3GiB");
}
};
- SegmentLoaderConfig segmentLoaderConfig = new SegmentLoaderConfig()
- {
- @Override
- public boolean isVirtualStorage()
- {
- return true;
- }
- };
+ SegmentLoaderConfig segmentLoaderConfig =
SegmentLoaderConfig.builder().virtualStorage(true).build();
DruidServerConfig druidServerConfig = new DruidServerConfig(runtimeInfo,
segmentLoaderConfig);
Assert.assertEquals(HumanReadableBytes.parse("45TiB"),
druidServerConfig.getMaxSize());
}
@@ -132,7 +125,7 @@ public class DruidServerConfigTest
mapper.setInjectableValues(new
InjectableValues.Std().addValue(ObjectMapper.class, new DefaultObjectMapper())
.addValue(
SegmentLoaderConfig.class,
- new
SegmentLoaderConfig().setLocations(locations)
+
SegmentLoaderConfig.builder().locations(locations).build()
)
.addValue(RuntimeInfo.class, new RuntimeInfo()));
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 06c01e1ad5a..f45d49a7a2c 100644
--- a/server/src/test/java/org/apache/druid/guice/StorageNodeModuleTest.java
+++ b/server/src/test/java/org/apache/druid/guice/StorageNodeModuleTest.java
@@ -174,9 +174,9 @@ public class StorageNodeModuleTest
{
// 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));
+ // The provider forces virtual storage via
config.toEphemeralVirtualStorage() regardless of the node's flag.
+ Mockito.when(segmentLoaderConfig.toEphemeralVirtualStorage())
+
.thenReturn(SegmentLoaderConfig.builder().virtualStorage(true).build());
final StorageLoadingThreadPool pool = injector().getInstance(
Key.get(StorageLoadingThreadPool.class, EphemeralStorageLoading.class)
diff --git
a/server/src/test/java/org/apache/druid/segment/loading/PermitLimitedSegmentRangeReaderTest.java
b/server/src/test/java/org/apache/druid/segment/loading/PermitLimitedSegmentRangeReaderTest.java
index 193d8474b1f..c725ec8db4d 100644
---
a/server/src/test/java/org/apache/druid/segment/loading/PermitLimitedSegmentRangeReaderTest.java
+++
b/server/src/test/java/org/apache/druid/segment/loading/PermitLimitedSegmentRangeReaderTest.java
@@ -31,14 +31,7 @@ class PermitLimitedSegmentRangeReaderTest
private static StorageLoadingThreadPool oneThreadPool()
{
return StorageLoadingThreadPool.createFromConfig(
- new SegmentLoaderConfig()
- {
- @Override
- public int getVirtualStorageLoadThreads()
- {
- return 1;
- }
- }.setVirtualStorage(true)
+
SegmentLoaderConfig.builder().virtualStorageLoadThreads(1).virtualStorage(true).build()
);
}
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 a3db293b6ad..eb520c2169a 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
@@ -29,18 +29,16 @@ import org.junit.jupiter.api.Test;
class SegmentLoaderConfigTest
{
@Test
- void testSetVirtualStorage()
+ void testBuilderVirtualStorageFlags()
{
- final SegmentLoaderConfig config = new SegmentLoaderConfig();
-
- // Verify default values
- Assertions.assertFalse(config.isVirtualStorage());
- Assertions.assertFalse(config.isVirtualStorageEphemeral());
-
- // Set both to true
- config.setVirtualStorage(true).setVirtualStorageIsEphemeral(true);
-
- // Verify both fields are set
+ // Defaults are false.
+ final SegmentLoaderConfig defaults = SegmentLoaderConfig.builder().build();
+ Assertions.assertFalse(defaults.isVirtualStorage());
+ Assertions.assertFalse(defaults.isVirtualStorageEphemeral());
+
+ // The builder sets both.
+ final SegmentLoaderConfig config =
+
SegmentLoaderConfig.builder().virtualStorage(true).virtualStorageIsEphemeral(true).build();
Assertions.assertTrue(config.isVirtualStorage());
Assertions.assertTrue(config.isVirtualStorageEphemeral());
}
@@ -82,15 +80,16 @@ class SegmentLoaderConfigTest
}
@Test
- public void testWithVirtualStorageReturnsCopyAndDoesNotMutateOriginal()
+ public void
testToEphemeralVirtualStorageReturnsCopyAndDoesNotMutateOriginal()
{
- final SegmentLoaderConfig original = new SegmentLoaderConfig();
+ final SegmentLoaderConfig original = SegmentLoaderConfig.builder().build();
Assertions.assertFalse(original.isVirtualStorage());
- final SegmentLoaderConfig copy = original.withVirtualStorage(true);
+ final SegmentLoaderConfig copy = original.toEphemeralVirtualStorage();
- // The copy has the flag flipped, while other settings are preserved.
+ // The copy is an ephemeral virtual-storage config, while tuning settings
are preserved.
Assertions.assertTrue(copy.isVirtualStorage());
+ Assertions.assertTrue(copy.isVirtualStorageEphemeral());
Assertions.assertEquals(original.getVirtualStorageLoadThreads(),
copy.getVirtualStorageLoadThreads());
Assertions.assertEquals(original.isVirtualStorageUseVirtualThreads(),
copy.isVirtualStorageUseVirtualThreads());
@@ -98,4 +97,43 @@ class SegmentLoaderConfigTest
Assertions.assertNotSame(original, copy);
Assertions.assertFalse(original.isVirtualStorage());
}
+
+ @Test
+ void testEnablingVirtualStorageClearsIncompatibleSettingsButKeepsTuning()
throws Exception
+ {
+ final ObjectMapper jsonMapper = new DefaultObjectMapper();
+ jsonMapper.setInjectableValues(new
InjectableValues.Std().addValue(RuntimeInfo.class, new RuntimeInfo()));
+ final SegmentLoaderConfig node = jsonMapper.readValue(
+ "{\"lazyLoadOnStart\": true,"
+ + " \"deleteOnRemove\": false,"
+ + " \"infoDir\": \"/var/druid/segment-cache/info_dir\","
+ + " \"numThreadsToLoadSegmentsIntoPageCacheOnDownload\": 4,"
+ + " \"numThreadsToLoadSegmentsIntoPageCacheOnBootstrap\": 2,"
+ + " \"virtualStorageMetadataReservationEstimate\": 99999999,"
+ + " \"virtualStorageCoalesceGapBytes\": 65536}",
+ SegmentLoaderConfig.class
+ );
+
+ final SegmentLoaderConfig virtual = node.toEphemeralVirtualStorage();
+
+ // Classic on-disk-cache settings are reset to virtual-storage-safe
values...
+ Assertions.assertFalse(virtual.isLazyLoadOnStart());
+ Assertions.assertTrue(virtual.isDeleteOnRemove());
+ Assertions.assertNull(virtual.getInfoDir());
+ Assertions.assertEquals(0,
virtual.getNumThreadsToLoadSegmentsIntoPageCacheOnDownload());
+ Assertions.assertEquals(0,
virtual.getNumThreadsToLoadSegmentsIntoPageCacheOnBootstrap());
+ // ...while virtual-storage tuning is preserved.
+ Assertions.assertTrue(virtual.isVirtualStorage());
+ Assertions.assertEquals(99999999L,
virtual.getVirtualStorageMetadataReservationEstimate());
+ Assertions.assertEquals(65536L,
virtual.getVirtualStorageCoalesceGapBytes());
+
+ // The original node config is untouched, and a plain toBuilder copy
preserves everything (only the ephemeral
+ // virtual-storage derivation sanitizes).
+ Assertions.assertTrue(node.isLazyLoadOnStart());
+ Assertions.assertEquals(4,
node.getNumThreadsToLoadSegmentsIntoPageCacheOnDownload());
+ final SegmentLoaderConfig plainCopy = node.toBuilder().build();
+ Assertions.assertTrue(plainCopy.isLazyLoadOnStart());
+ Assertions.assertFalse(plainCopy.isDeleteOnRemove());
+ Assertions.assertEquals(4,
plainCopy.getNumThreadsToLoadSegmentsIntoPageCacheOnDownload());
+ }
}
diff --git
a/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerConcurrencyTest.java
b/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerConcurrencyTest.java
index 5c2b482779c..921d1597cc2 100644
---
a/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerConcurrencyTest.java
+++
b/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerConcurrencyTest.java
@@ -136,46 +136,16 @@ class SegmentLocalCacheManagerConcurrencyTest
locations.add(locationConfig);
locations.add(locationConfig2);
- loaderConfig = new SegmentLoaderConfig()
- {
- @Override
- public List<StorageLocationConfig> getLocations()
- {
- return locations;
- }
-
- @Override
- public File getInfoDir()
- {
- return new File(tempDir, "info");
- }
- };
- vsfLoaderConfig = new SegmentLoaderConfig()
- {
- @Override
- public List<StorageLocationConfig> getLocations()
- {
- return locations;
- }
-
- @Override
- public boolean isVirtualStorage()
- {
- return true;
- }
-
- @Override
- public int getVirtualStorageLoadThreads()
- {
- return Runtime.getRuntime().availableProcessors();
- }
-
- @Override
- public File getInfoDir()
- {
- return new File(tempDir, "info");
- }
- };
+ loaderConfig = SegmentLoaderConfig.builder()
+ .locations(locations)
+ .infoDir(new File(tempDir, "info"))
+ .build();
+ vsfLoaderConfig = SegmentLoaderConfig.builder()
+ .locations(locations)
+ .virtualStorage(true)
+
.virtualStorageLoadThreads(Runtime.getRuntime().availableProcessors())
+ .infoDir(new File(tempDir, "info"))
+ .build();
final List<StorageLocation> storageLocations =
loaderConfig.toStorageLocations();
location = storageLocations.get(0);
location2 = storageLocations.get(1);
diff --git
a/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerPartialAcquireTest.java
b/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerPartialAcquireTest.java
index 7a6ed49c37e..7a823904db6 100644
---
a/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerPartialAcquireTest.java
+++
b/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerPartialAcquireTest.java
@@ -253,10 +253,11 @@ class SegmentLocalCacheManagerPartialAcquireTest
FileUtils.mkdirp(cacheRoot);
final StorageLocationConfig locConfig = new
StorageLocationConfig(cacheRoot, 1024L * 1024L * 1024L, null);
- final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig()
- .setLocations(List.of(locConfig))
- .setVirtualStorage(true)
- .setVirtualStoragePartialDownloadsEnabled(true);
+ final SegmentLoaderConfig loaderConfig = SegmentLoaderConfig.builder()
+ .locations(locConfig)
+ .virtualStorage(true)
+ .virtualStoragePartialDownloadsEnabled(true)
+ .build();
final List<StorageLocation> storageLocations =
loaderConfig.toStorageLocations();
manager = new SegmentLocalCacheManager(
storageLocations,
@@ -350,7 +351,7 @@ class SegmentLocalCacheManagerPartialAcquireTest
throws ExecutionException, InterruptedException, IOException
{
final StorageLocation loc = manager.getLocations().get(0);
- final long estimate = new
SegmentLoaderConfig().getVirtualStorageMetadataReservationEstimate();
+ final long estimate =
SegmentLoaderConfig.builder().build().getVirtualStorageMetadataReservationEstimate();
// A lazy (PARTIAL) acquire mounts only the metadata entry (downloads the
header); bundles stay lazy until a cursor
// is built, so exactly one weak reservation + one weak load is expected.
@@ -630,10 +631,11 @@ class SegmentLocalCacheManagerPartialAcquireTest
final File plainCacheRoot = new File(perTestTempDir, "plain_cache_" +
ThreadLocalRandom.current().nextInt(Integer.MAX_VALUE));
FileUtils.mkdirp(plainCacheRoot);
final StorageLocationConfig locConfig = new
StorageLocationConfig(plainCacheRoot, 1024L * 1024L * 1024L, null);
- final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig()
- .setLocations(List.of(locConfig))
- .setVirtualStorage(true)
- .setVirtualStoragePartialDownloadsEnabled(true);
+ final SegmentLoaderConfig loaderConfig = SegmentLoaderConfig.builder()
+ .locations(locConfig)
+ .virtualStorage(true)
+ .virtualStoragePartialDownloadsEnabled(true)
+ .build();
final List<StorageLocation> storageLocations =
loaderConfig.toStorageLocations();
final SegmentLocalCacheManager plain = new SegmentLocalCacheManager(
storageLocations,
@@ -697,10 +699,11 @@ class SegmentLocalCacheManagerPartialAcquireTest
// eager counterparts: the returned segment is an eager
QueryableIndexSegment (NOT PartialQueryableIndexSegment),
// and no PartialSegmentMetadataCacheEntry is registered on the location.
final StorageLocationConfig locConfig = new
StorageLocationConfig(cacheRoot, 1024L * 1024L * 1024L, null);
- final SegmentLoaderConfig disabledConfig = new SegmentLoaderConfig()
- .setLocations(List.of(locConfig))
- .setVirtualStorage(true)
- .setVirtualStoragePartialDownloadsEnabled(false);
+ final SegmentLoaderConfig disabledConfig = SegmentLoaderConfig.builder()
+ .locations(locConfig)
+ .virtualStorage(true)
+ .virtualStoragePartialDownloadsEnabled(false)
+ .build();
final List<StorageLocation> storageLocations =
disabledConfig.toStorageLocations();
final SegmentLocalCacheManager disabledManager = new
SegmentLocalCacheManager(
storageLocations,
@@ -807,10 +810,11 @@ class SegmentLocalCacheManagerPartialAcquireTest
// A manager with partial downloads disabled (operator toggled the flag
off) over the same cache dir must reclaim
// the now-unusable partial layout at bootstrap rather than reserving it
and failing at query time.
final StorageLocationConfig locConfig = new
StorageLocationConfig(cacheRoot, 1024L * 1024L * 1024L, null);
- final SegmentLoaderConfig disabledConfig = new SegmentLoaderConfig()
- .setLocations(List.of(locConfig))
- .setVirtualStorage(true)
- .setVirtualStoragePartialDownloadsEnabled(false);
+ final SegmentLoaderConfig disabledConfig = SegmentLoaderConfig.builder()
+ .locations(locConfig)
+ .virtualStorage(true)
+ .virtualStoragePartialDownloadsEnabled(false)
+ .build();
final List<StorageLocation> storageLocations =
disabledConfig.toStorageLocations();
final SegmentLocalCacheManager disabledManager = new
SegmentLocalCacheManager(
storageLocations,
diff --git
a/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerPartialDropTest.java
b/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerPartialDropTest.java
index bc4901bcb33..f0f0d227846 100644
---
a/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerPartialDropTest.java
+++
b/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerPartialDropTest.java
@@ -156,7 +156,7 @@ class SegmentLocalCacheManagerPartialDropTest extends
InitializedNullHandlingTes
FileUtils.mkdirp(infoDir);
final StorageLocationConfig locConfig = new
StorageLocationConfig(cacheDir, ESTIMATE * 16, null);
- final SegmentLoaderConfig loaderConfig = new
SegmentLoaderConfig().setLocations(List.of(locConfig));
+ final SegmentLoaderConfig loaderConfig =
SegmentLoaderConfig.builder().locations(locConfig).build();
final List<StorageLocation> locations = loaderConfig.toStorageLocations();
manager = new SegmentLocalCacheManager(
locations,
diff --git
a/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerPartialRuleLoadTest.java
b/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerPartialRuleLoadTest.java
index f63563474e4..1d72d2a2ed4 100644
---
a/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerPartialRuleLoadTest.java
+++
b/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerPartialRuleLoadTest.java
@@ -710,10 +710,11 @@ class SegmentLocalCacheManagerPartialRuleLoadTest
final List<StorageLocationConfig> locConfigs = locationRoots.stream()
.map(root -> new StorageLocationConfig(root, 1024L * 1024L * 1024L,
null))
.toList();
- final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig()
- .setLocations(locConfigs)
- .setVirtualStorage(virtualStorage)
- .setVirtualStoragePartialDownloadsEnabled(partialDownloadsEnabled);
+ final SegmentLoaderConfig loaderConfig = SegmentLoaderConfig.builder()
+ .locations(locConfigs)
+ .virtualStorage(virtualStorage)
+ .virtualStoragePartialDownloadsEnabled(partialDownloadsEnabled)
+ .build();
final List<StorageLocation> storageLocations =
loaderConfig.toStorageLocations();
return new SegmentLocalCacheManager(
storageLocations,
diff --git
a/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerTest.java
b/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerTest.java
index d6f96739fc2..458da17f219 100644
---
a/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerTest.java
+++
b/server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerTest.java
@@ -118,16 +118,10 @@ public class SegmentLocalCacheManagerTest extends
InitializedNullHandlingTest
{
// Only injecting config locations without locations shouldn't really be
the case.
// It possibly suggests an issue with injection.
- final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig()
- {
- @Override
- public List<StorageLocationConfig> getLocations()
- {
- return Collections.singletonList(
- new StorageLocationConfig(localSegmentCacheDir, null, null)
- );
- }
- };
+ final SegmentLoaderConfig loaderConfig =
+ SegmentLoaderConfig.builder()
+ .locations(new
StorageLocationConfig(localSegmentCacheDir, null, null))
+ .build();
manager = new SegmentLocalCacheManager(
ImmutableList.of(),
@@ -146,7 +140,7 @@ public class SegmentLocalCacheManagerTest extends
InitializedNullHandlingTest
final ImmutableList<StorageLocation> locations = ImmutableList.of(
new StorageLocation(localSegmentCacheDir, 10000000000L, null)
);
- final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig();
+ final SegmentLoaderConfig loaderConfig =
SegmentLoaderConfig.builder().build();
SegmentLocalCacheManager manager = new SegmentLocalCacheManager(
locations,
loaderConfig,
@@ -161,7 +155,7 @@ public class SegmentLocalCacheManagerTest extends
InitializedNullHandlingTest
@Test
public void testCanHandleSegmentsWithEmptyLocationsAndConfigLocations()
{
- final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig();
+ final SegmentLoaderConfig loaderConfig =
SegmentLoaderConfig.builder().build();
SegmentLocalCacheManager manager = new SegmentLocalCacheManager(
ImmutableList.of(),
loaderConfig,
@@ -176,7 +170,7 @@ public class SegmentLocalCacheManagerTest extends
InitializedNullHandlingTest
@Test
public void testGetCachedSegmentsWhenCanHandleSegmentsIsFalse()
{
- final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig();
+ final SegmentLoaderConfig loaderConfig =
SegmentLoaderConfig.builder().build();
SegmentLocalCacheManager manager = new SegmentLocalCacheManager(
null,
loaderConfig,
@@ -346,24 +340,13 @@ public class SegmentLocalCacheManagerTest extends
InitializedNullHandlingTest
}
@Test
- public void testLoadSegmentInPageCache() throws IOException,
SegmentLoadingException
+ public void testLoadSegmentInPageCache() throws IOException
{
- final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig()
- {
- @Override
- public int getNumThreadsToLoadSegmentsIntoPageCacheOnDownload()
- {
- return 1;
- }
-
- @Override
- public List<StorageLocationConfig> getLocations()
- {
- return Collections.singletonList(
- new StorageLocationConfig(localSegmentCacheDir, null, null)
- );
- }
- };
+ final SegmentLoaderConfig loaderConfig =
+ SegmentLoaderConfig.builder()
+ .numThreadsToLoadSegmentsIntoPageCacheOnDownload(1)
+ .locations(new
StorageLocationConfig(localSegmentCacheDir, null, null))
+ .build();
final List<StorageLocation> locations = loaderConfig.toStorageLocations();
SegmentLocalCacheManager manager = new SegmentLocalCacheManager(
locations,
@@ -415,7 +398,7 @@ public class SegmentLocalCacheManagerTest extends
InitializedNullHandlingTest
final StorageLocationConfig locationConfig2 = new
StorageLocationConfig(localStorageFolder2, 1000000000L, null);
locations.add(locationConfig2);
- final SegmentLoaderConfig loaderConfig = new
SegmentLoaderConfig().setLocations(locations);
+ final SegmentLoaderConfig loaderConfig =
SegmentLoaderConfig.builder().locations(locations).build();
final List<StorageLocation> storageLocations =
loaderConfig.toStorageLocations();
SegmentLocalCacheManager manager = new SegmentLocalCacheManager(
storageLocations,
@@ -454,7 +437,7 @@ public class SegmentLocalCacheManagerTest extends
InitializedNullHandlingTest
final StorageLocationConfig locationConfig2 = new
StorageLocationConfig(localStorageFolder2, 10000000L, null);
locations.add(locationConfig2);
- final SegmentLoaderConfig loaderConfig = new
SegmentLoaderConfig().setLocations(locations);
+ final SegmentLoaderConfig loaderConfig =
SegmentLoaderConfig.builder().locations(locations).build();
final List<StorageLocation> storageLocations =
loaderConfig.toStorageLocations();
SegmentLocalCacheManager manager = new SegmentLocalCacheManager(
storageLocations,
@@ -495,7 +478,7 @@ public class SegmentLocalCacheManagerTest extends
InitializedNullHandlingTest
final StorageLocationConfig locationConfig2 = new
StorageLocationConfig(localStorageFolder2, 10000000L, null);
locations.add(locationConfig2);
- final SegmentLoaderConfig loaderConfig = new
SegmentLoaderConfig().setLocations(locations);
+ final SegmentLoaderConfig loaderConfig =
SegmentLoaderConfig.builder().locations(locations).build();
final List<StorageLocation> storageLocations =
loaderConfig.toStorageLocations();
SegmentLocalCacheManager manager = new SegmentLocalCacheManager(
storageLocations,
@@ -535,7 +518,7 @@ public class SegmentLocalCacheManagerTest extends
InitializedNullHandlingTest
final StorageLocationConfig locationConfig2 = new
StorageLocationConfig(localStorageFolder2, 10L, null);
locations.add(locationConfig2);
- final SegmentLoaderConfig loaderConfig = new
SegmentLoaderConfig().setLocations(locations);
+ final SegmentLoaderConfig loaderConfig =
SegmentLoaderConfig.builder().locations(locations).build();
final List<StorageLocation> storageLocations =
loaderConfig.toStorageLocations();
SegmentLocalCacheManager manager = new SegmentLocalCacheManager(
storageLocations,
@@ -592,7 +575,7 @@ public class SegmentLocalCacheManagerTest extends
InitializedNullHandlingTest
);
}
- final SegmentLoaderConfig loaderConfig = new
SegmentLoaderConfig().setLocations(locationConfigs);
+ final SegmentLoaderConfig loaderConfig =
SegmentLoaderConfig.builder().locations(locationConfigs).build();
SegmentLocalCacheManager manager = new SegmentLocalCacheManager(
locations,
loaderConfig,
@@ -670,11 +653,11 @@ public class SegmentLocalCacheManagerTest extends
InitializedNullHandlingTest
locations.add(locationConfig2);
locations.add(locationConfig3);
- final SegmentLoaderConfig loaderConfig = new
SegmentLoaderConfig().setLocations(locations);
+ final SegmentLoaderConfig loaderConfig =
SegmentLoaderConfig.builder().locations(locations).build();
final List<StorageLocation> storageLocations =
loaderConfig.toStorageLocations();
SegmentLocalCacheManager manager = new SegmentLocalCacheManager(
storageLocations,
- new SegmentLoaderConfig().setLocations(locations),
+ SegmentLoaderConfig.builder().locations(locations).build(),
StorageLoadingThreadPool.createFromConfig(loaderConfig),
new LeastBytesUsedStorageLocationSelectorStrategy(storageLocations),
TestHelper.getTestIndexIO(jsonMapper, ColumnConfig.DEFAULT),
@@ -752,7 +735,7 @@ public class SegmentLocalCacheManagerTest extends
InitializedNullHandlingTest
locationConfigs.add(locationConfig2);
locationConfigs.add(locationConfig3);
- SegmentLoaderConfig segmentLoaderConfig = new
SegmentLoaderConfig().setLocations(locationConfigs);
+ SegmentLoaderConfig segmentLoaderConfig =
SegmentLoaderConfig.builder().locations(locationConfigs).build();
final List<StorageLocation> locations =
segmentLoaderConfig.toStorageLocations();
SegmentLocalCacheManager manager = new SegmentLocalCacheManager(
@@ -835,7 +818,7 @@ public class SegmentLocalCacheManagerTest extends
InitializedNullHandlingTest
public void testGetBootstrapSegment() throws SegmentLoadingException
{
final StorageLocationConfig locationConfig = new
StorageLocationConfig(localSegmentCacheDir, 10000L, null);
- final SegmentLoaderConfig loaderConfig = new
SegmentLoaderConfig().setLocations(ImmutableList.of(locationConfig));
+ final SegmentLoaderConfig loaderConfig =
SegmentLoaderConfig.builder().locations(locationConfig).build();
final List<StorageLocation> storageLocations =
loaderConfig.toStorageLocations();
SegmentLocalCacheManager manager = new SegmentLocalCacheManager(
storageLocations,
@@ -860,20 +843,10 @@ public class SegmentLocalCacheManagerTest extends
InitializedNullHandlingTest
public void testGetBootstrapSegmentLazy() throws SegmentLoadingException
{
final StorageLocationConfig locationConfig = new
StorageLocationConfig(localSegmentCacheDir, 10000L, null);
- final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig()
- {
- @Override
- public boolean isLazyLoadOnStart()
- {
- return true;
- }
-
- @Override
- public List<StorageLocationConfig> getLocations()
- {
- return List.of(locationConfig);
- }
- };
+ final SegmentLoaderConfig loaderConfig = SegmentLoaderConfig.builder()
+
.lazyLoadOnStart(true)
+
.locations(locationConfig)
+ .build();
final List<StorageLocation> storageLocations =
loaderConfig.toStorageLocations();
SegmentLocalCacheManager manager = new SegmentLocalCacheManager(
storageLocations,
@@ -897,31 +870,11 @@ public class SegmentLocalCacheManagerTest extends
InitializedNullHandlingTest
public void testGetSegmentVirtualStorage() throws Exception
{
final StorageLocationConfig locationConfig = new
StorageLocationConfig(localSegmentCacheDir, 10000L, null);
- final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig()
- {
- @Override
- public List<StorageLocationConfig> getLocations()
- {
- return ImmutableList.of(locationConfig);
- }
-
- @Override
- public boolean isVirtualStorage()
- {
- return true;
- }
-
- @Override
- public File getInfoDir()
- {
- try {
- return tmpFolder.newFolder();
- }
- catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
- };
+ final SegmentLoaderConfig loaderConfig = SegmentLoaderConfig.builder()
+
.locations(locationConfig)
+
.virtualStorage(true)
+
.infoDir(tmpFolder.newFolder())
+ .build();
final List<StorageLocation> storageLocations =
loaderConfig.toStorageLocations();
SegmentLocalCacheManager manager = new SegmentLocalCacheManager(
storageLocations,
@@ -969,26 +922,11 @@ public class SegmentLocalCacheManagerTest extends
InitializedNullHandlingTest
public void testVirtualStorageRejectsNonPositiveLoadThreads()
{
final StorageLocationConfig locationConfig = new
StorageLocationConfig(localSegmentCacheDir, 10000L, null);
- final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig()
- {
- @Override
- public List<StorageLocationConfig> getLocations()
- {
- return ImmutableList.of(locationConfig);
- }
-
- @Override
- public boolean isVirtualStorage()
- {
- return true;
- }
-
- @Override
- public int getVirtualStorageLoadThreads()
- {
- return 0;
- }
- };
+ final SegmentLoaderConfig loaderConfig = SegmentLoaderConfig.builder()
+
.locations(locationConfig)
+
.virtualStorage(true)
+
.virtualStorageLoadThreads(0)
+ .build();
final List<StorageLocation> storageLocations =
loaderConfig.toStorageLocations();
MatcherAssert.assertThat(
Assert.assertThrows(
@@ -1014,31 +952,11 @@ public class SegmentLocalCacheManagerTest extends
InitializedNullHandlingTest
public void testGetBootstrapSegmentVirtualStorage() throws Exception
{
final StorageLocationConfig locationConfig = new
StorageLocationConfig(localSegmentCacheDir, 10000L, null);
- final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig()
- {
- @Override
- public List<StorageLocationConfig> getLocations()
- {
- return ImmutableList.of(locationConfig);
- }
-
- @Override
- public boolean isVirtualStorage()
- {
- return true;
- }
-
- @Override
- public File getInfoDir()
- {
- try {
- return tmpFolder.newFolder();
- }
- catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
- };
+ final SegmentLoaderConfig loaderConfig = SegmentLoaderConfig.builder()
+
.locations(locationConfig)
+
.virtualStorage(true)
+
.infoDir(tmpFolder.newFolder())
+ .build();
final List<StorageLocation> storageLocations =
loaderConfig.toStorageLocations();
SegmentLocalCacheManager manager = new SegmentLocalCacheManager(
storageLocations,
@@ -1087,20 +1005,10 @@ public class SegmentLocalCacheManagerTest extends
InitializedNullHandlingTest
public void testGetBootstrapSegmentVirtualStorageSegmentAlreadyCached()
throws Exception
{
final StorageLocationConfig locationConfig = new
StorageLocationConfig(localSegmentCacheDir, 10000L, null);
- final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig()
- {
- @Override
- public List<StorageLocationConfig> getLocations()
- {
- return ImmutableList.of(locationConfig);
- }
-
- @Override
- public boolean isVirtualStorage()
- {
- return true;
- }
- };
+ final SegmentLoaderConfig loaderConfig = SegmentLoaderConfig.builder()
+
.locations(locationConfig)
+
.virtualStorage(true)
+ .build();
final List<StorageLocation> storageLocations =
loaderConfig.toStorageLocations();
SegmentLocalCacheManager manager = new SegmentLocalCacheManager(
storageLocations,
@@ -1165,31 +1073,11 @@ public class SegmentLocalCacheManagerTest extends
InitializedNullHandlingTest
public void testGetSegmentVirtualStorageMountAfterDrop() throws Exception
{
final StorageLocationConfig locationConfig = new
StorageLocationConfig(localSegmentCacheDir, 10L, null);
- final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig()
- {
- @Override
- public List<StorageLocationConfig> getLocations()
- {
- return ImmutableList.of(locationConfig);
- }
-
- @Override
- public boolean isVirtualStorage()
- {
- return true;
- }
-
- @Override
- public File getInfoDir()
- {
- try {
- return tmpFolder.newFolder();
- }
- catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
- };
+ final SegmentLoaderConfig loaderConfig = SegmentLoaderConfig.builder()
+
.locations(locationConfig)
+
.virtualStorage(true)
+
.infoDir(tmpFolder.newFolder())
+ .build();
final List<StorageLocation> storageLocations =
loaderConfig.toStorageLocations();
SegmentLocalCacheManager manager = new SegmentLocalCacheManager(
storageLocations,
@@ -1249,32 +1137,12 @@ public class SegmentLocalCacheManagerTest extends
InitializedNullHandlingTest
{
final StorageLocationConfig locationConfig = new
StorageLocationConfig(localSegmentCacheDir, 10000L, null);
final File infoDir = tmpFolder.newFolder();
- final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig()
- {
- @Override
- public List<StorageLocationConfig> getLocations()
- {
- return ImmutableList.of(locationConfig);
- }
-
- @Override
- public boolean isVirtualStorage()
- {
- return true;
- }
-
- @Override
- public boolean isVirtualStorageEphemeral()
- {
- return true;
- }
-
- @Override
- public File getInfoDir()
- {
- return infoDir;
- }
- };
+ final SegmentLoaderConfig loaderConfig = SegmentLoaderConfig.builder()
+
.locations(locationConfig)
+
.virtualStorage(true)
+
.virtualStorageIsEphemeral(true)
+
.infoDir(infoDir)
+ .build();
final List<StorageLocation> storageLocations =
loaderConfig.toStorageLocations();
final SegmentLocalCacheManager manager = new SegmentLocalCacheManager(
storageLocations,
@@ -1399,7 +1267,7 @@ public class SegmentLocalCacheManagerTest extends
InitializedNullHandlingTest
final List<StorageLocationConfig> locationConfigs = new ArrayList<>();
final StorageLocationConfig locationConfig = new
StorageLocationConfig(localSegmentCacheDir, 10000000000L, null);
locationConfigs.add(locationConfig);
- final SegmentLoaderConfig loaderConfig = new
SegmentLoaderConfig().setLocations(locationConfigs);
+ final SegmentLoaderConfig loaderConfig =
SegmentLoaderConfig.builder().locations(locationConfigs).build();
final List<StorageLocation> locations = loaderConfig.toStorageLocations();
return new SegmentLocalCacheManager(
locations,
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
index 196b1d96437..3f92a401f34 100644
---
a/server/src/test/java/org/apache/druid/segment/loading/StorageLoadingThreadPoolTest.java
+++
b/server/src/test/java/org/apache/druid/segment/loading/StorageLoadingThreadPoolTest.java
@@ -28,46 +28,32 @@ class StorageLoadingThreadPoolTest
{
private static SegmentLoaderConfig oneVirtualThreadConfig()
{
- return new SegmentLoaderConfig()
- {
- @Override
- public int getVirtualStorageLoadThreads()
- {
- return 1;
- }
- }.setVirtualStorage(true);
+ return
SegmentLoaderConfig.builder().virtualStorageLoadThreads(1).virtualStorage(true).build();
}
private static SegmentLoaderConfig fixedThreadConfig()
{
- return new SegmentLoaderConfig()
- {
- @Override
- public int getVirtualStorageLoadThreads()
- {
- return 2;
- }
-
- @Override
- public boolean isVirtualStorageUseVirtualThreads()
- {
- return false;
- }
- }.setVirtualStorage(true);
+ return SegmentLoaderConfig.builder()
+ .virtualStorageLoadThreads(2)
+ .virtualStorageUseVirtualThreads(false)
+ .virtualStorage(true)
+ .build();
}
@Test
void testCreateFromConfigIsUnavailableWhenNotVirtualStorage()
{
- Assertions.assertFalse(StorageLoadingThreadPool.createFromConfig(new
SegmentLoaderConfig()).isAvailable());
+ Assertions.assertFalse(
+
StorageLoadingThreadPool.createFromConfig(SegmentLoaderConfig.builder().build()).isAvailable()
+ );
}
@Test
void testCreateFromConfigIsAvailableWhenVirtualStorage()
{
- // The shared ephemeral pool is built this way:
createFromConfig(config.withVirtualStorage(true)).
+ // The shared ephemeral pool is built this way:
createFromConfig(config.toEphemeralVirtualStorage()).
final StorageLoadingThreadPool pool =
- StorageLoadingThreadPool.createFromConfig(new
SegmentLoaderConfig().withVirtualStorage(true));
+
StorageLoadingThreadPool.createFromConfig(SegmentLoaderConfig.builder().build().toEphemeralVirtualStorage());
try {
Assertions.assertTrue(pool.isAvailable());
Assertions.assertNotNull(pool.getExecutorService());
@@ -80,14 +66,8 @@ class StorageLoadingThreadPoolTest
@Test
void testCreateFromConfigRejectsNonPositiveThreadCountInVirtualStorage()
{
- final SegmentLoaderConfig config = new SegmentLoaderConfig()
- {
- @Override
- public int getVirtualStorageLoadThreads()
- {
- return 0;
- }
- }.setVirtualStorage(true);
+ final SegmentLoaderConfig config =
+
SegmentLoaderConfig.builder().virtualStorageLoadThreads(0).virtualStorage(true).build();
Assertions.assertThrows(DruidException.class, () ->
StorageLoadingThreadPool.createFromConfig(config));
}
diff --git
a/server/src/test/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderatorTester.java
b/server/src/test/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderatorTester.java
index 93571eba5ba..76344d9565d 100644
---
a/server/src/test/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderatorTester.java
+++
b/server/src/test/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderatorTester.java
@@ -233,14 +233,9 @@ public class StreamAppenderatorTester implements
AutoCloseable
taskIntervalUnlocker
);
} else {
- SegmentLoaderConfig segmentLoaderConfig = new SegmentLoaderConfig()
- {
- @Override
- public int getDropSegmentDelayMillis()
- {
- return delayInMilli;
- }
- };
+ SegmentLoaderConfig segmentLoaderConfig = SegmentLoaderConfig.builder()
+
.dropSegmentDelayMillis(delayInMilli)
+ .build();
appenderator = Appenderators.createRealtime(
segmentLoaderConfig,
schema.getDataSource(),
diff --git
a/server/src/test/java/org/apache/druid/server/SegmentManagerTest.java
b/server/src/test/java/org/apache/druid/server/SegmentManagerTest.java
index ea0412709a9..22c20731c11 100644
--- a/server/src/test/java/org/apache/druid/server/SegmentManagerTest.java
+++ b/server/src/test/java/org/apache/druid/server/SegmentManagerTest.java
@@ -70,7 +70,6 @@ import org.junit.rules.TemporaryFolder;
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
-import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -105,50 +104,21 @@ public class SegmentManagerTest extends
InitializedNullHandlingTest
{
EmittingLogger.registerEmitter(new NoopServiceEmitter());
final File segmentCacheDir = temporaryFolder.newFolder();
- final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig()
- {
- @Override
- public File getInfoDir()
- {
- return segmentCacheDir;
- }
-
- @Override
- public List<StorageLocationConfig> getLocations()
- {
- return Collections.singletonList(
- new StorageLocationConfig(segmentCacheDir, null, null)
- );
- }
- };
+ final SegmentLoaderConfig loaderConfig = SegmentLoaderConfig.builder()
+ .infoDir(segmentCacheDir)
+ .locations(new StorageLocationConfig(segmentCacheDir, null, null))
+ .build();
final File vsfRoot = temporaryFolder.newFolder();
final File virtualSegmentCacheDir = new File(vsfRoot, "segmentCache");
FileUtils.mkdirp(virtualSegmentCacheDir);
final File vsfInfoDir = new File(vsfRoot, "info");
FileUtils.mkdirp(vsfInfoDir);
- final SegmentLoaderConfig virtualLoaderConfig = new SegmentLoaderConfig()
- {
- @Override
- public File getInfoDir()
- {
- return vsfInfoDir;
- }
-
- @Override
- public List<StorageLocationConfig> getLocations()
- {
- return Collections.singletonList(
- new StorageLocationConfig(virtualSegmentCacheDir, null, null)
- );
- }
-
- @Override
- public boolean isVirtualStorage()
- {
- return true;
- }
- };
+ final SegmentLoaderConfig virtualLoaderConfig =
SegmentLoaderConfig.builder()
+ .infoDir(vsfInfoDir)
+ .locations(new StorageLocationConfig(virtualSegmentCacheDir, null,
null))
+ .virtualStorage(true)
+ .build();
final ObjectMapper objectMapper = TestHelper.makeJsonMapper();
objectMapper.registerSubtypes(TestSegmentUtils.TestLoadSpec.class);
diff --git
a/server/src/test/java/org/apache/druid/server/StatusResourceTest.java
b/server/src/test/java/org/apache/druid/server/StatusResourceTest.java
index 54dab17a40a..c677d48ef28 100644
--- a/server/src/test/java/org/apache/druid/server/StatusResourceTest.java
+++ b/server/src/test/java/org/apache/druid/server/StatusResourceTest.java
@@ -28,6 +28,7 @@ import org.apache.druid.guice.StartupInjectorBuilder;
import org.apache.druid.guice.TestDruidModule;
import org.apache.druid.initialization.DruidModule;
import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.segment.loading.SegmentLoaderConfig;
import org.apache.druid.utils.JvmUtils;
import org.junit.Assert;
import org.junit.Test;
@@ -103,7 +104,10 @@ public class StatusResourceTest
private void testHiddenPropertiesWithPropertyFileName(String fileName)
throws Exception
{
Injector injector = new StartupInjectorBuilder()
- .add(new PropertiesModule(Collections.singletonList(fileName)))
+ .add(
+ new PropertiesModule(Collections.singletonList(fileName)),
+ binder ->
binder.bind(SegmentLoaderConfig.class).toInstance(SegmentLoaderConfig.builder().build())
+ )
.build();
Map<String, String> returnedProperties =
injector.getInstance(StatusResource.class).getProperties();
Set<String> lowerCasePropertyNames = returnedProperties.keySet()
diff --git
a/server/src/test/java/org/apache/druid/server/coordination/SegmentCacheBootstrapperCacheTest.java
b/server/src/test/java/org/apache/druid/server/coordination/SegmentCacheBootstrapperCacheTest.java
index 7878630d2af..e5c4063eac6 100644
---
a/server/src/test/java/org/apache/druid/server/coordination/SegmentCacheBootstrapperCacheTest.java
+++
b/server/src/test/java/org/apache/druid/server/coordination/SegmentCacheBootstrapperCacheTest.java
@@ -46,7 +46,6 @@ import org.junit.rules.TemporaryFolder;
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
-import java.util.Collections;
import java.util.List;
/**
@@ -75,22 +74,10 @@ public class SegmentCacheBootstrapperCacheTest
{
infoDir = temporaryFolder.newFolder();
cacheDir = temporaryFolder.newFolder();
- loaderConfig = new SegmentLoaderConfig()
- {
- @Override
- public File getInfoDir()
- {
- return infoDir;
- }
-
- @Override
- public List<StorageLocationConfig> getLocations()
- {
- return Collections.singletonList(
- new StorageLocationConfig(cacheDir, MAX_SIZE, null)
- );
- }
- };
+ loaderConfig = SegmentLoaderConfig.builder()
+ .infoDir(infoDir)
+ .locations(new StorageLocationConfig(cacheDir, MAX_SIZE, null))
+ .build();
objectMapper = TestHelper.makeJsonMapper();
objectMapper.registerSubtypes(TestSegmentUtils.TestLoadSpec.class);
@@ -116,7 +103,7 @@ public class SegmentCacheBootstrapperCacheTest
public void testLoadStartStopWithEmptyLocations() throws IOException
{
final List<StorageLocation> emptyLocations = ImmutableList.of();
- final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig();
+ final SegmentLoaderConfig loaderConfig =
SegmentLoaderConfig.builder().build();
segmentManager = new SegmentManager(
new SegmentLocalCacheManager(
emptyLocations,
diff --git
a/server/src/test/java/org/apache/druid/server/coordination/SegmentCacheBootstrapperTest.java
b/server/src/test/java/org/apache/druid/server/coordination/SegmentCacheBootstrapperTest.java
index 594554022c7..75f09988314 100644
---
a/server/src/test/java/org/apache/druid/server/coordination/SegmentCacheBootstrapperTest.java
+++
b/server/src/test/java/org/apache/druid/server/coordination/SegmentCacheBootstrapperTest.java
@@ -40,7 +40,6 @@ import org.junit.rules.TemporaryFolder;
import java.io.File;
import java.io.IOException;
-import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
@@ -65,34 +64,12 @@ public class SegmentCacheBootstrapperTest
final File segmentCacheDir = temporaryFolder.newFolder();
segmentAnnouncer = new TestDataSegmentAnnouncer();
- segmentLoaderConfig = new SegmentLoaderConfig()
- {
- @Override
- public File getInfoDir()
- {
- return segmentCacheDir;
- }
-
- @Override
- public int getNumLoadingThreads()
- {
- return 5;
- }
-
- @Override
- public int getAnnounceIntervalMillis()
- {
- return 50;
- }
-
- @Override
- public List<StorageLocationConfig> getLocations()
- {
- return Collections.singletonList(
- new StorageLocationConfig(segmentCacheDir, null, null)
- );
- }
- };
+ segmentLoaderConfig = SegmentLoaderConfig.builder()
+ .infoDir(segmentCacheDir)
+ .numLoadingThreads(5)
+ .announceIntervalMillis(50)
+ .locations(new StorageLocationConfig(segmentCacheDir, null, null))
+ .build();
coordinatorClient = new TestCoordinatorClient();
serviceEmitter = new StubServiceEmitter();
diff --git
a/server/src/test/java/org/apache/druid/server/coordination/SegmentLoadDropHandlerTest.java
b/server/src/test/java/org/apache/druid/server/coordination/SegmentLoadDropHandlerTest.java
index 4c8fc0c4079..b46150423aa 100644
---
a/server/src/test/java/org/apache/druid/server/coordination/SegmentLoadDropHandlerTest.java
+++
b/server/src/test/java/org/apache/druid/server/coordination/SegmentLoadDropHandlerTest.java
@@ -44,7 +44,6 @@ import org.mockito.Mockito;
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
-import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -72,40 +71,13 @@ public class SegmentLoadDropHandlerTest
scheduledRunnable = new ArrayList<>();
segmentAnnouncer = new TestDataSegmentAnnouncer();
- segmentLoaderConfig = new SegmentLoaderConfig()
- {
- @Override
- public File getInfoDir()
- {
- return segmentCacheDir;
- }
-
- @Override
- public int getNumLoadingThreads()
- {
- return 5;
- }
-
- @Override
- public int getAnnounceIntervalMillis()
- {
- return 50;
- }
-
- @Override
- public List<StorageLocationConfig> getLocations()
- {
- return Collections.singletonList(
- new StorageLocationConfig(segmentCacheDir, null, null)
- );
- }
-
- @Override
- public int getDropSegmentDelayMillis()
- {
- return 0;
- }
- };
+ segmentLoaderConfig = SegmentLoaderConfig.builder()
+ .infoDir(segmentCacheDir)
+ .numLoadingThreads(5)
+ .announceIntervalMillis(50)
+ .locations(new StorageLocationConfig(segmentCacheDir, null, null))
+ .dropSegmentDelayMillis(0)
+ .build();
scheduledExecutorFactory = (corePoolSize, nameFormat) -> {
// Override normal behavior by adding the runnable to a list so that you
can make sure
@@ -298,40 +270,13 @@ public class SegmentLoadDropHandlerTest
Mockito.doNothing().when(segmentManager).dropSegment(ArgumentMatchers.any());
final File storageDir = temporaryFolder.newFolder();
- final SegmentLoaderConfig noAnnouncerSegmentLoaderConfig = new
SegmentLoaderConfig()
- {
- @Override
- public File getInfoDir()
- {
- return storageDir;
- }
-
- @Override
- public int getNumLoadingThreads()
- {
- return 5;
- }
-
- @Override
- public int getAnnounceIntervalMillis()
- {
- return 0;
- }
-
- @Override
- public List<StorageLocationConfig> getLocations()
- {
- return Collections.singletonList(
- new StorageLocationConfig(storageDir, null, null)
- );
- }
-
- @Override
- public int getDropSegmentDelayMillis()
- {
- return 0;
- }
- };
+ final SegmentLoaderConfig noAnnouncerSegmentLoaderConfig =
SegmentLoaderConfig.builder()
+ .infoDir(storageDir)
+ .numLoadingThreads(5)
+ .announceIntervalMillis(0)
+ .locations(new StorageLocationConfig(storageDir, null, null))
+ .dropSegmentDelayMillis(0)
+ .build();
final SegmentLoadDropHandler handler = initSegmentLoadDropHandler(
noAnnouncerSegmentLoaderConfig,
diff --git
a/server/src/test/java/org/apache/druid/server/coordination/SegmentManagerBroadcastJoinIndexedTableTest.java
b/server/src/test/java/org/apache/druid/server/coordination/SegmentManagerBroadcastJoinIndexedTableTest.java
index 5ee311c389b..a16e24bfc0e 100644
---
a/server/src/test/java/org/apache/druid/server/coordination/SegmentManagerBroadcastJoinIndexedTableTest.java
+++
b/server/src/test/java/org/apache/druid/server/coordination/SegmentManagerBroadcastJoinIndexedTableTest.java
@@ -129,22 +129,10 @@ public class SegmentManagerBroadcastJoinIndexedTableTest
extends InitializedNull
segmentCacheDir = temporaryFolder.newFolder();
segmentDeepStorageDir = temporaryFolder.newFolder();
- final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig()
- {
- @Override
- public File getInfoDir()
- {
- return segmentCacheDir;
- }
-
- @Override
- public List<StorageLocationConfig> getLocations()
- {
- return Collections.singletonList(
- new StorageLocationConfig(segmentCacheDir, null, null)
- );
- }
- };
+ final SegmentLoaderConfig loaderConfig = SegmentLoaderConfig.builder()
+ .infoDir(segmentCacheDir)
+ .locations(new StorageLocationConfig(segmentCacheDir, null, null))
+ .build();
final List<StorageLocation> storageLocations =
loaderConfig.toStorageLocations();
segmentCacheManager = new SegmentLocalCacheManager(
diff --git
a/server/src/test/java/org/apache/druid/server/coordination/SegmentManagerThreadSafetyTest.java
b/server/src/test/java/org/apache/druid/server/coordination/SegmentManagerThreadSafetyTest.java
index 75f9c347a9c..fec53f4c20f 100644
---
a/server/src/test/java/org/apache/druid/server/coordination/SegmentManagerThreadSafetyTest.java
+++
b/server/src/test/java/org/apache/druid/server/coordination/SegmentManagerThreadSafetyTest.java
@@ -100,22 +100,10 @@ public class SegmentManagerThreadSafetyTest
segmentCacheDir = temporaryFolder.newFolder();
segmentDeepStorageDir = temporaryFolder.newFolder();
- final SegmentLoaderConfig loaderConfig = new SegmentLoaderConfig()
- {
- @Override
- public File getInfoDir()
- {
- return segmentCacheDir;
- }
-
- @Override
- public List<StorageLocationConfig> getLocations()
- {
- return Collections.singletonList(
- new StorageLocationConfig(segmentCacheDir, null, null)
- );
- }
- };
+ final SegmentLoaderConfig loaderConfig = SegmentLoaderConfig.builder()
+ .infoDir(segmentCacheDir)
+ .locations(new StorageLocationConfig(segmentCacheDir, null, null))
+ .build();
final List<StorageLocation> storageLocations =
loaderConfig.toStorageLocations();
segmentCacheManager = new SegmentLocalCacheManager(
storageLocations,
diff --git
a/server/src/test/java/org/apache/druid/server/metrics/SegmentStatsMonitorTest.java
b/server/src/test/java/org/apache/druid/server/metrics/SegmentStatsMonitorTest.java
index 373b1c1dd2d..5aa4f926da2 100644
---
a/server/src/test/java/org/apache/druid/server/metrics/SegmentStatsMonitorTest.java
+++
b/server/src/test/java/org/apache/druid/server/metrics/SegmentStatsMonitorTest.java
@@ -52,7 +52,7 @@ public class SegmentStatsMonitorTest
private SegmentLoadDropHandler segmentLoadDropMgr;
private ServiceEmitter serviceEmitter;
private SegmentStatsMonitor monitor;
- private final SegmentLoaderConfig segmentLoaderConfig = new
SegmentLoaderConfig();
+ private final SegmentLoaderConfig segmentLoaderConfig =
SegmentLoaderConfig.builder().build();
@Before
public void setUp()
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
