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 c70397cbcb7 fix: partial entry ephemeral cache release (#19872)
c70397cbcb7 is described below
commit c70397cbcb744fd7400aa124be9b3b4ae046f4d5
Author: Clint Wylie <[email protected]>
AuthorDate: Tue Aug 4 13:32:42 2026 -0700
fix: partial entry ephemeral cache release (#19872)
---
.../EmbeddedMSQProjectionPartialDownloadsTest.java | 190 +++++++++++++++++++++
.../druid/segment/loading/StorageLocation.java | 46 ++++-
.../druid/segment/loading/StorageLocationTest.java | 150 ++++++++++++++++
3 files changed, 380 insertions(+), 6 deletions(-)
diff --git
a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/msq/EmbeddedMSQProjectionPartialDownloadsTest.java
b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/msq/EmbeddedMSQProjectionPartialDownloadsTest.java
new file mode 100644
index 00000000000..a76c8517ef8
--- /dev/null
+++
b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/msq/EmbeddedMSQProjectionPartialDownloadsTest.java
@@ -0,0 +1,190 @@
+/*
+ * 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.AggregateProjectionSpec;
+import org.apache.druid.data.input.impl.LongDimensionSchema;
+import org.apache.druid.data.input.impl.StringDimensionSchema;
+import org.apache.druid.data.input.impl.TimestampSpec;
+import org.apache.druid.indexer.granularity.UniformGranularitySpec;
+import org.apache.druid.indexing.common.task.TaskBuilder;
+import
org.apache.druid.indexing.common.task.batch.parallel.ParallelIndexSupervisorTask;
+import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.java.util.common.granularity.Granularities;
+import org.apache.druid.query.QueryContexts;
+import org.apache.druid.query.aggregation.LongSumAggregatorFactory;
+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.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.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * End-to-end test that ingests a projection segment and runs the exact GROUP
BY the projection serves through an
+ * MSQ task, so its GroupBy leaf processor acquires the projection bundle
(dependent on {@code __base}) under
+ * {@code AcquireMode.PARTIAL} and releases it at cursor close under the
ephemeral cache.
+ */
+class EmbeddedMSQProjectionPartialDownloadsTest extends EmbeddedClusterTestBase
+{
+ private static final String PROJECTION_NAME = "country_delta";
+ private static final long MONITOR_QUIESCE_TIMEOUT_MILLIS = 3_000L;
+ private static final String SELECT_SQL = "SELECT \"countryName\",
SUM(\"delta\") AS s FROM \"%s\" GROUP BY \"countryName\"";
+
+ 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)
+ .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 = "projection-partial-" + IdUtils.getRandomId();
+ ingestSegmentWithProjection();
+ }
+
+ @Override
+ protected void refreshDatasourceName()
+ {
+ // Keep the datasource stable: it is ingested once in loadData() before
all tests.
+ }
+
+ @Test
+ void testProjectionServedGroupByDoesNotCrashWorker()
+ {
+ final LatchableEmitter emitter = indexer.latchableEmitter();
+
+ emitter.awaitMetricQuiescent(StorageMonitor.VSF_READ_COUNT,
MONITOR_QUIESCE_TIMEOUT_MILLIS);
+ emitter.flush();
+
+ final SqlTaskStatus status = msqApis.submitTaskSql(
+ Map.of(QueryContexts.USE_PROJECTION, PROJECTION_NAME),
+ SELECT_SQL,
+ dataSource
+ );
+ cluster.callApi().waitForTaskToSucceed(status.getTaskId(), overlord);
+
+ // Confirm the partial-load path actually engaged: on-demand deep-storage
range reads. The worker's StorageMonitor
+ // only emits while the ephemeral task is alive, so the events land during
the query; waitForEventAggregate also
+ // considers events already emitted since the last flush and fails the
test on timeout if none ever arrived.
+ emitter.waitForEventAggregate(
+ matcher -> matcher.hasMetricName(StorageMonitor.VSF_READ_COUNT),
+ aggregate -> aggregate.hasSumAtLeast(1L)
+ );
+ }
+
+ private void ingestSegmentWithProjection() throws IOException
+ {
+ final File tmpDir = cluster.getTestFolder().newFolder();
+ final File inputFile = new File(tmpDir, "projection-input.json");
+ final String inputData =
+
"{\"time\":\"2024-01-01T00:10:00Z\",\"channel\":\"#en\",\"countryName\":\"US\",\"delta\":10}\n"
+ +
"{\"time\":\"2024-01-01T00:20:00Z\",\"channel\":\"#en\",\"countryName\":\"US\",\"delta\":5}\n"
+ +
"{\"time\":\"2024-01-01T00:30:00Z\",\"channel\":\"#en\",\"countryName\":\"CA\",\"delta\":3}\n"
+ +
"{\"time\":\"2024-01-01T00:40:00Z\",\"channel\":\"#fr\",\"countryName\":\"FR\",\"delta\":7}\n"
+ +
"{\"time\":\"2024-01-01T00:50:00Z\",\"channel\":\"#fr\",\"countryName\":\"US\",\"delta\":2}\n";
+ Files.write(inputFile.toPath(),
inputData.getBytes(StandardCharsets.UTF_8));
+
+ final AggregateProjectionSpec projection =
+ AggregateProjectionSpec.builder(PROJECTION_NAME)
+ .groupingColumns(new
StringDimensionSchema("countryName"))
+ .aggregators(new
LongSumAggregatorFactory("sumDelta", "delta"))
+ .build();
+
+ final UniformGranularitySpec granularitySpec = new UniformGranularitySpec(
+ Granularities.HOUR,
+ Granularities.NONE,
+ false,
+ List.of(Intervals.of("2024-01-01/2024-01-02"))
+ );
+
+ final String taskId = IdUtils.getRandomId();
+ final ParallelIndexSupervisorTask task = TaskBuilder
+ .ofTypeIndexParallel()
+ .jsonInputFormat()
+ .localInputSourceWithFiles(inputFile)
+ .dataSchema(
+ builder -> builder
+ .withDataSource(dataSource)
+ .withTimestamp(new TimestampSpec("time", "iso", null))
+ .withGranularity(granularitySpec)
+ .withDimensions(
+ new StringDimensionSchema("channel"),
+ new StringDimensionSchema("countryName"),
+ new LongDimensionSchema("delta")
+ )
+ .withProjections(List.of(projection))
+ )
+ .tuningConfig(t -> t.withMaxNumConcurrentSubTasks(1))
+ .withId(taskId);
+
+ cluster.callApi().onLeaderOverlord(o -> o.runTask(taskId, task));
+ cluster.callApi().waitForTaskToSucceed(taskId, overlord);
+ cluster.callApi().waitForAllSegmentsToBeAvailable(dataSource, coordinator,
broker);
+ }
+}
diff --git
a/server/src/main/java/org/apache/druid/segment/loading/StorageLocation.java
b/server/src/main/java/org/apache/druid/segment/loading/StorageLocation.java
index 49fc135a9a8..6330f20cb4c 100644
--- a/server/src/main/java/org/apache/druid/segment/loading/StorageLocation.java
+++ b/server/src/main/java/org/apache/druid/segment/loading/StorageLocation.java
@@ -536,6 +536,7 @@ public class StorageLocation
{
lock.writeLock().lock();
try {
+ final WeakCacheEntry[] evicted = new WeakCacheEntry[1];
weakCacheEntries.computeIfPresent(
id,
(cacheEntryIdentifier, weakCacheEntry) -> {
@@ -544,19 +545,40 @@ public class StorageLocation
}
final boolean isMounted = weakCacheEntry.cacheEntry.isMounted();
unlinkWeakEntry(weakCacheEntry);
- weakCacheEntry.unmount();
if (isMounted) {
weakStats.getAndUpdate(s ->
s.evict(weakCacheEntry.cacheEntry.getSize()));
}
+ evicted[0] = weakCacheEntry;
return null;
}
);
+ unmountEvictedWeakEntry(evicted[0]);
}
finally {
lock.writeLock().unlock();
}
}
+ /**
+ * Fire {@link WeakCacheEntry#unmount()} for an entry that has just been
removed from {@link #weakCacheEntries}.
+ * <p>
+ * MUST be called <em>after</em> the enclosing {@link
Map#computeIfPresent}/{@link Map#computeIfAbsent} callback has
+ * returned, never from inside it. Unmounting a partial bundle entry drains
its {@link Phaser}, whose termination
+ * cascades into {@code PartialSegmentBundleCacheEntry.doActualUnmount} ->
parent bundle hold release ->
+ * {@code weakCacheEntries.computeIfPresent} for a <em>different</em> key.
Structurally modifying
+ * {@link #weakCacheEntries} from within another entry's compute callback
trips HashMap's fail-fast check
+ * (a {@link java.util.ConcurrentModificationException}). Removing the entry
first, then unmounting, keeps the
+ * cascade's map mutations sequential. Callers hold the write lock across
both steps so no reserve/re-mount can race
+ * the not-yet-drained phaser.
+ */
+ @GuardedBy("lock")
+ private void unmountEvictedWeakEntry(@Nullable WeakCacheEntry evicted)
+ {
+ if (evicted != null) {
+ evicted.unmount();
+ }
+ }
+
/**
* Creates a release runnable for a {@link WeakCacheEntry} that handles
immediate eviction when configured.
* If {@link #areWeakEntriesEphemeral} is true and there are no more holds
after releasing, the entry is immediately
@@ -578,6 +600,7 @@ public class StorageLocation
lock.writeLock().lock();
try {
+ final WeakCacheEntry[] evicted = new WeakCacheEntry[1];
weakCacheEntries.computeIfPresent(
weakEntry.cacheEntry.getId(),
(cacheEntryIdentifier, weakCacheEntry) -> {
@@ -587,16 +610,17 @@ public class StorageLocation
if ((isNewEntry && !isMounted)
|| (areWeakEntriesEphemeral && !weakCacheEntry.isHeld())) {
unlinkWeakEntry(weakCacheEntry);
- weakCacheEntry.unmount(); // call even if never mounted, to
terminate the phaser
if (isMounted) {
weakStats.getAndUpdate(s ->
s.evict(weakCacheEntry.cacheEntry.getSize()));
}
+ evicted[0] = weakCacheEntry;
return null;
} else {
return weakCacheEntry;
}
}
);
+ unmountEvictedWeakEntry(evicted[0]);
}
finally {
lock.writeLock().unlock();
@@ -739,6 +763,7 @@ public class StorageLocation
private void unmountReclaimed(ReclaimResult reclaimResult)
{
if (reclaimResult != null) {
+ final List<WeakCacheEntry> toUnmount = new ArrayList<>();
for (WeakCacheEntry removed : reclaimResult.getEvictions()) {
weakCacheEntries.computeIfAbsent(
removed.cacheEntry.getId(),
@@ -747,13 +772,16 @@ public class StorageLocation
weakStats.getAndUpdate(s ->
s.evict(removed.cacheEntry.getSize()));
// .. but make sure the same identifier wasn't moved to a static
load before we actually unmount it
if (!staticCacheEntries.containsKey(cacheEntryIdentifier)) {
- removed.unmount();
+ toUnmount.add(removed);
weakStats.getAndUpdate(WeakStats::unmount);
}
return null;
}
);
}
+ for (WeakCacheEntry removed : toUnmount) {
+ unmountEvictedWeakEntry(removed);
+ }
}
}
@@ -839,11 +867,17 @@ public class StorageLocation
entry.unmount();
}
staticCacheEntries.clear();
- while (head != null) {
- head.unmount();
- head = head.next;
+ final List<WeakCacheEntry> weakToUnmount = new ArrayList<>();
+ for (WeakCacheEntry entry = head; entry != null; entry = entry.next) {
+ weakToUnmount.add(entry);
}
+ head = null;
+ tail = null;
+ hand = null;
weakCacheEntries.clear();
+ for (WeakCacheEntry entry : weakToUnmount) {
+ entry.unmount();
+ }
}
finally {
lock.writeLock().unlock();
diff --git
a/server/src/test/java/org/apache/druid/segment/loading/StorageLocationTest.java
b/server/src/test/java/org/apache/druid/segment/loading/StorageLocationTest.java
index 433a5b414bd..b3b0f32d914 100644
---
a/server/src/test/java/org/apache/druid/segment/loading/StorageLocationTest.java
+++
b/server/src/test/java/org/apache/druid/segment/loading/StorageLocationTest.java
@@ -33,6 +33,8 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import javax.annotation.Nullable;
+import java.io.Closeable;
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
@@ -582,6 +584,93 @@ class StorageLocationTest
Assertions.assertEquals(0, location.getWeakStats().getHoldBytes());
}
+ @Test
+ public void
testEphemeralWeakEntryUnmountCascadeDoesNotThrowConcurrentModification()
+ {
+ final StorageLocation location = new StorageLocation(tempDir, 10_000L,
null);
+ location.setAreWeakEntriesEphemeral(true);
+
+ // Parent entry (e.g. a __base bundle), held by the child for its lifetime
and evicted only when the child unmounts.
+ final CascadingUnmountCacheEntry parent = new
CascadingUnmountCacheEntry("parent", 100L, null);
+ final StorageLocation.ReservationHold<?> parentHold =
+ location.addWeakReservationHold(parent.getId(), () -> parent);
+ Assertions.assertNotNull(parentHold);
+ parent.mount(location);
+
+ // Child entry (a partial bundle) whose unmount() releases the parent's
cache hold, re-entering StorageLocation.
+ final CascadingUnmountCacheEntry child = new
CascadingUnmountCacheEntry("child", 100L, parentHold);
+ final StorageLocation.ReservationHold<?> childHold =
+ location.addWeakReservationHold(child.getId(), () -> child);
+ Assertions.assertNotNull(childHold);
+ child.mount(location);
+
+ // Releasing the child's last hold evicts it (ephemeral) and fires
child.unmount() -> parentHold.close() -> parent
+ // eviction, all under the location write lock. Before the fix this threw
ConcurrentModificationException.
+ Assertions.assertDoesNotThrow(childHold::close);
+
+ Assertions.assertTrue(child.wasUnmounted());
+ Assertions.assertTrue(parent.wasUnmounted());
+ Assertions.assertNull(location.getCacheEntry(child.getId()));
+ Assertions.assertNull(location.getCacheEntry(parent.getId()));
+ }
+
+ @Test
+ public void
testRemoveUnheldWeakEntryUnmountCascadeDoesNotThrowConcurrentModification()
+ {
+ final StorageLocation location = new StorageLocation(tempDir, 10_000L,
null);
+
+ // The parent is deliberately NOT mounted: on a non-ephemeral location,
releasing the child's hold on it removes it
+ // via the `isNewEntry && !isMounted` branch of
createWeakEntryReleaseRunnable, and that removal (re-entering
+ // weakCacheEntries for the parent key) is what reproduces the cascade. Do
NOT add parent.mount() here — a mounted,
+ // non-ephemeral parent is never removed on hold release, so no nested map
mutation occurs and the test would pass
+ // even against the buggy code.
+ final CascadingUnmountCacheEntry parent = new
CascadingUnmountCacheEntry("parent", 100L, null);
+ final StorageLocation.ReservationHold<?> parentHold =
+ location.addWeakReservationHold(parent.getId(), () -> parent);
+ Assertions.assertNotNull(parentHold);
+
+ // Child registered WITHOUT a hold (the bootstrap reserveWeak path that
removeUnheldWeakEntry cleans up).
+ final CascadingUnmountCacheEntry child = new
CascadingUnmountCacheEntry("child", 100L, parentHold);
+ Assertions.assertTrue(location.reserveWeak(child));
+ child.mount(location);
+
+ Assertions.assertDoesNotThrow(() ->
location.removeUnheldWeakEntry(child.getId()));
+
+ Assertions.assertTrue(child.wasUnmounted());
+ Assertions.assertTrue(parent.wasUnmounted());
+ Assertions.assertNull(location.getCacheEntry(child.getId()));
+ Assertions.assertNull(location.getCacheEntry(parent.getId()));
+ }
+
+ @Test
+ public void testReclaimUnmountCascadeDoesNotThrowConcurrentModification()
+ {
+ // Small location so the filler reservation forces reclaim of the child
(parent is held, so reclaim skips it).
+ final StorageLocation location = new StorageLocation(tempDir, 100L, null);
+
+ // As in testRemoveUnheld..., the parent is deliberately NOT mounted so
that releasing the child's hold on it
+ // removes it (the `isNewEntry && !isMounted` branch), driving the nested
weakCacheEntries mutation. Adding
+ // parent.mount() would leave the parent in the map on release and defang
the reproduction.
+ final CascadingUnmountCacheEntry parent = new
CascadingUnmountCacheEntry("parent", 40L, null);
+ final StorageLocation.ReservationHold<?> parentHold =
+ location.addWeakReservationHold(parent.getId(), () -> parent);
+ Assertions.assertNotNull(parentHold);
+
+ // Child registered unheld and mounted, sole holder of the parent's cache
hold; it is the reclaim target.
+ final CascadingUnmountCacheEntry child = new
CascadingUnmountCacheEntry("child", 40L, parentHold);
+ Assertions.assertTrue(location.reserveWeak(child));
+ child.mount(location);
+
+ final CascadingUnmountCacheEntry filler = new
CascadingUnmountCacheEntry("filler", 40L, null);
+ Assertions.assertDoesNotThrow(() -> location.reserveWeak(filler));
+
+ Assertions.assertTrue(child.wasUnmounted());
+ Assertions.assertTrue(parent.wasUnmounted());
+ Assertions.assertNull(location.getCacheEntry(child.getId()));
+ Assertions.assertNull(location.getCacheEntry(parent.getId()));
+ Assertions.assertTrue(location.isWeakReserved(filler.getId()));
+ }
+
@SuppressWarnings({"GuardedBy", "FieldAccessNotGuarded"})
private void verifyLoc(long maxSize, StorageLocation loc)
{
@@ -759,6 +848,67 @@ class StorageLocationTest
}
}
+ /**
+ * A {@link CacheEntry} whose {@link #unmount()} optionally closes another
{@link Closeable} (e.g. a parent bundle's
+ * {@link StorageLocation.ReservationHold}), reproducing the partial-load
cascade where unmounting one weak entry
+ * re-enters {@link StorageLocation} to release a hold on a different weak
entry.
+ */
+ private static final class CascadingUnmountCacheEntry implements CacheEntry
+ {
+ private final StringCacheIdentifier id;
+ private final long size;
+ @Nullable
+ private final Closeable onUnmount;
+ private boolean mounted = false;
+ private boolean unmounted = false;
+
+ private CascadingUnmountCacheEntry(String id, long size, @Nullable
Closeable onUnmount)
+ {
+ this.id = new StringCacheIdentifier(id);
+ this.size = size;
+ this.onUnmount = onUnmount;
+ }
+
+ @Override
+ public StringCacheIdentifier getId()
+ {
+ return id;
+ }
+
+ @Override
+ public long getSize()
+ {
+ return size;
+ }
+
+ @Override
+ public boolean isMounted()
+ {
+ return mounted;
+ }
+
+ @Override
+ public void mount(StorageLocation location)
+ {
+ mounted = true;
+ }
+
+ @Override
+ public void unmount()
+ {
+ unmounted = true;
+ mounted = false;
+ if (onUnmount != null) {
+ CloseableUtils.closeAndWrapExceptions(onUnmount);
+ }
+ }
+
+ private boolean wasUnmounted()
+ {
+ return unmounted;
+ }
+ }
+
private static final class TestResizableCacheEntry implements
ResizableCacheEntry
{
private final StringCacheIdentifier id;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
