github-advanced-security[bot] commented on code in PR #20073:
URL: https://github.com/apache/druid/pull/20073#discussion_r3847182158
##########
server/src/test/java/org/apache/druid/segment/loading/SegmentLocalCacheManagerPartialAcquireTest.java:
##########
@@ -962,6 +965,188 @@
);
}
+ @Test
+ void testAbandonedAcquireLeavesAConcurrentAcquiresEntryAlone()
+ throws ExecutionException, InterruptedException, IOException
+ {
+ final StorageLocation location = manager.getLocations().get(0);
+ final SegmentCacheEntryIdentifier id = new
SegmentCacheEntryIdentifier(SEGMENT_ID);
+
+ // Two acquires holding the same freshly reserved entry, neither of which
resolves its future, so nothing mounts
+ // it. The acquire that created the entry gives up first - a canceled or
timed out query, say.
+ final PartialSegmentMetadataCacheEntry entry;
+ final Closer acquires = Closer.create();
+ try {
+ // The first acquire reserves the entry; the second joins it. Both are
registered with the closer so a failed
+ // assertion cannot leak a hold into the shared manager fixture;
AcquireSegmentAction.close() is idempotent, so
+ // closing the first one early below is safe.
+ final AcquireSegmentAction creator = acquires.register(
+ manager.acquireSegment(partialSegment, AcquireMode.PARTIAL)
+ );
+ acquires.register(manager.acquireSegment(partialSegment,
AcquireMode.PARTIAL));
+ entry =
Assertions.assertInstanceOf(PartialSegmentMetadataCacheEntry.class,
location.getCacheEntry(id));
+ Assertions.assertFalse(entry.isMounted());
+
+ creator.close();
+ Assertions.assertSame(
+ entry,
+ location.getCacheEntry(id),
+ "the entry the second acquire is holding must survive the first
giving up"
+ );
+ }
+ finally {
+ acquires.close();
+ }
+ // The second acquire letting go does not remove it either - removal is
the creating hold's job - so it is left
+ // registered and unmounted, reclaimable, and reusable by the next acquire.
+ Assertions.assertSame(entry, location.getCacheEntry(id));
+ Assertions.assertFalse(entry.isMounted());
+
+ // A later acquire mounts that same entry and serves the segment from it.
+ try (AcquireSegmentAction action = manager.acquireSegment(partialSegment,
AcquireMode.PARTIAL)) {
+ final AcquireSegmentResult result = action.getSegmentFuture().get();
+ try (Segment segment =
result.getReferenceProvider().acquireReference().orElseThrow()) {
+ Assertions.assertEquals(SEGMENT_ID, segment.getId());
+ final TimeBoundaryInspector inspector =
segment.as(TimeBoundaryInspector.class);
+ Assertions.assertNotNull(inspector);
+ Assertions.assertEquals(TIME, inspector.getMinTime());
+ Assertions.assertEquals(TIME.plusMinutes(3), inspector.getMaxTime());
+ }
+ Assertions.assertSame(entry, location.getCacheEntry(id));
+ Assertions.assertTrue(entry.isMounted());
+ }
+ }
+
+ @Test
+ void testAbandonedAcquireWhoseMountFailsResolvesToAnUnavailableSegmentToo()
throws Exception
+ {
+ // Deep storage that can produce a range reader now but not serve a read
later: openRangeReader() checks the V10
+ // file at acquire time, and the test deletes it before the load task gets
to run.
+ final File vanishingStorage =
temporaryFolder.newFolder("vanishing_storage");
+ final File v10File = new File(vanishingStorage, IndexIO.V10_FILE_NAME);
+ Files.copy(new File(DEEP_STORAGE_DIR, IndexIO.V10_FILE_NAME).toPath(),
v10File.toPath());
+ final DataSegment vanishingSegment = DataSegment.builder(SEGMENT_ID)
+
.shardSpec(NoneShardSpec.instance())
+ .loadSpec(Map.of("type",
"local", "path", vanishingStorage.getAbsolutePath()))
+ .size(0)
+ .build();
+
+ final File gatedCacheRoot =
temporaryFolder.newFolder("gated_cache_mount_failure");
+ final SegmentLoaderConfig gatedConfig = SegmentLoaderConfig.builder()
+ .locations(new StorageLocationConfig(gatedCacheRoot, 1024L * 1024L *
1024L, null))
+ .virtualStorage(true)
+ .virtualStoragePartialDownloadsEnabled(true)
+ .virtualStorageUseVirtualThreads(false)
+ .virtualStorageLoadThreads(1)
+ .build();
+ final List<StorageLocation> storageLocations =
gatedConfig.toStorageLocations();
+ final StorageLoadingThreadPool gatedPool =
StorageLoadingThreadPool.createFromConfig(gatedConfig);
+ final SegmentLocalCacheManager gatedManager = new SegmentLocalCacheManager(
+ storageLocations,
+ gatedConfig,
+ gatedPool,
+ new LeastBytesUsedStorageLocationSelectorStrategy(storageLocations),
+ TestHelper.getTestIndexIO(jsonMapper, ColumnConfig.DEFAULT),
+ jsonMapper
+ );
+
+ final CountDownLatch atGate = new CountDownLatch(1);
+ final CountDownLatch openGate = new CountDownLatch(1);
+ try {
+ // (intentionally unused) local so errorprone's CheckReturnValue is
satisfied
+ @SuppressWarnings("unused")
+ ListenableFuture<?> unused = gatedPool.getExecutorService().submit(() ->
{
+ atGate.countDown();
+ return openGate.await(30, TimeUnit.SECONDS);
+ });
+ Assertions.assertTrue(atGate.await(30, TimeUnit.SECONDS), "loading
thread must reach the gate");
+
+ final ListenableFuture<AcquireSegmentResult> future;
+ try (AcquireSegmentAction action =
gatedManager.acquireSegment(vanishingSegment, AcquireMode.PARTIAL)) {
+ future = action.getSegmentFuture();
+ }
+ // The mount will now fail on its header read rather than rolling back
cleanly, so the loss surfaces from
+ // mount() instead of from the pin - which is no reason to fail a query
that has already given up.
+ Assertions.assertTrue(v10File.delete(), "test needs the deep-storage
file gone before the mount runs");
+ openGate.countDown();
+
+ final AcquireSegmentResult result = future.get(30, TimeUnit.SECONDS);
+ Assertions.assertTrue(
+ result.getReferenceProvider().acquireReference().isEmpty(),
+ "an abandoned acquire whose mount failed must resolve to an
unavailable segment"
+ );
+ }
+ finally {
+ openGate.countDown();
+ gatedManager.drop(vanishingSegment);
+ gatedManager.shutdown();
+ gatedPool.stop();
+ }
+ }
+
+ @Test
+ void testAbandonedAcquireResolvesToAnUnavailableSegmentRatherThanFailing()
throws Exception
+ {
+ // One fixed loading thread, so the test can hold the load task at a gate
while it abandons the acquire.
+ final File gatedCacheRoot = temporaryFolder.newFolder("gated_cache");
+ final SegmentLoaderConfig gatedConfig = SegmentLoaderConfig.builder()
+ .locations(new StorageLocationConfig(gatedCacheRoot, 1024L * 1024L *
1024L, null))
+ .virtualStorage(true)
+ .virtualStoragePartialDownloadsEnabled(true)
+ .virtualStorageUseVirtualThreads(false)
+ .virtualStorageLoadThreads(1)
+ .build();
+ final List<StorageLocation> storageLocations =
gatedConfig.toStorageLocations();
+ final StorageLoadingThreadPool gatedPool =
StorageLoadingThreadPool.createFromConfig(gatedConfig);
+ final SegmentLocalCacheManager gatedManager = new SegmentLocalCacheManager(
+ storageLocations,
+ gatedConfig,
+ gatedPool,
+ new LeastBytesUsedStorageLocationSelectorStrategy(storageLocations),
+ TestHelper.getTestIndexIO(jsonMapper, ColumnConfig.DEFAULT),
+ jsonMapper
+ );
+
+ final CountDownLatch atGate = new CountDownLatch(1);
+ final CountDownLatch openGate = new CountDownLatch(1);
+ try {
+ // (intentionally unused) local so errorprone's CheckReturnValue is
satisfied
+ @SuppressWarnings("unused")
+ ListenableFuture<?> unused = gatedPool.getExecutorService().submit(() ->
{
+ atGate.countDown();
+ return openGate.await(30, TimeUnit.SECONDS);
+ });
Review Comment:
## CodeQL / Unread local variable
Variable 'ListenableFuture<?> unused' is never read.
[Show more
details](https://github.com/apache/druid/security/code-scanning/11910)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
