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 899f765a39a minor: remove AsyncResources#fromFutureCloseable (#20356)
899f765a39a is described below
commit 899f765a39af93bcb4e428529b0d67aaad598059
Author: Clint Wylie <[email protected]>
AuthorDate: Tue Sep 15 14:14:27 2026 -0700
minor: remove AsyncResources#fromFutureCloseable (#20356)
---
.../druid/common/asyncresource/AsyncResources.java | 50 +-----------
.../common/asyncresource/AsyncResourcesTest.java | 60 --------------
.../segment/loading/StorageLoadingThreadPool.java | 23 +-----
.../StorageLocationVirtualStorageManager.java | 22 +++++-
.../StorageLocationVirtualStorageManagerTest.java | 91 ++++++++++++++++++++++
5 files changed, 115 insertions(+), 131 deletions(-)
diff --git
a/processing/src/main/java/org/apache/druid/common/asyncresource/AsyncResources.java
b/processing/src/main/java/org/apache/druid/common/asyncresource/AsyncResources.java
index a7832937106..d4cf84ffa6c 100644
---
a/processing/src/main/java/org/apache/druid/common/asyncresource/AsyncResources.java
+++
b/processing/src/main/java/org/apache/druid/common/asyncresource/AsyncResources.java
@@ -24,7 +24,6 @@ import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
import org.apache.druid.collections.ResourceHolder;
-import org.apache.druid.utils.CloseableUtils;
import java.io.Closeable;
import java.util.List;
@@ -66,8 +65,8 @@ public class AsyncResources
* resource before the future completes cancels the future ({@code
cancel(true)}).
* <p>
* The result object must <b>not</b> have a lifecycle; closing the resource
does not close the result. Use this for
- * futures whose result is a plain value or a completion signal; use {@link
#fromFutureCloseable} or
- * {@link #ofCloseable} if the result has a lifecycle of its own.
+ * futures whose result is a plain value or a completion signal; use {@link
#ofCloseable} if the result has
+ * a lifecycle of its own.
*/
public static <T> AsyncResource<T> fromFutureUnmanaged(final
ListenableFuture<T> future)
{
@@ -94,51 +93,6 @@ public class AsyncResources
return retVal;
}
- /**
- * Returns an {@link AsyncResource} backed by a {@link ListenableFuture}
whose result <b>owns a lifecycle</b>: it
- * becomes ready when the future completes, exposing the result via {@link
AsyncResource#get()}, and the result is
- * managed as a {@link Closeable}. Closing the resource closes the result,
and a result that completes <i>after</i>
- * the resource was already closed is closed rather than leaked.
- *
- * <p>Closing the returned resource deliberately does <b>not</b> cancel the
future. Cancellation is what makes
- * futures-of-closeables unsafe in the first place: a task that produces its
value anyway hands it to a canceled
- * future, which drops it silently, and nothing is left to close it. Leaving
the future alone means the result
- * always arrives through the callback below, which closes it when the
resource is already gone. The cost is that
- * work already submitted runs to completion; a producer that wants real
cancellation should populate a
- * {@link SettableAsyncResource} itself and give it a {@link
SettableAsyncResource#setCanceler canceler} that can
- * abort safely.
- *
- * <p>This is the managed counterpart of {@link #fromFutureUnmanaged}; use
that for a future whose result is a plain
- * value or a completion signal with no lifecycle.
- */
- public static <T extends Closeable> AsyncResource<T>
fromFutureCloseable(final ListenableFuture<T> future)
- {
- final SettableAsyncResource<T> retVal = new SettableAsyncResource<>();
- Futures.addCallback(
- future,
- new FutureCallback<>()
- {
- @Override
- public void onSuccess(T result)
- {
- // Lost the race with close(): the resource is already closed, so
set() returns false and we own
- // closing the now-orphaned result.
- if (!retVal.set(ResourceHolder.fromCloseable(result))) {
- CloseableUtils.closeAndSuppressExceptions(result, ignored -> {});
- }
- }
-
- @Override
- public void onFailure(Throwable t)
- {
- retVal.setException(t);
- }
- },
- MoreExecutors.directExecutor()
- );
- return retVal;
- }
-
/**
* Returns an {@link AsyncResource} whose value is the result of calling
{@code function} on an underlying resource.
*
diff --git
a/processing/src/test/java/org/apache/druid/common/asyncresource/AsyncResourcesTest.java
b/processing/src/test/java/org/apache/druid/common/asyncresource/AsyncResourcesTest.java
index 4e036179308..a612cbf3366 100644
---
a/processing/src/test/java/org/apache/druid/common/asyncresource/AsyncResourcesTest.java
+++
b/processing/src/test/java/org/apache/druid/common/asyncresource/AsyncResourcesTest.java
@@ -24,9 +24,7 @@ import org.apache.druid.error.DruidException;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
-import java.io.Closeable;
import java.util.List;
-import java.util.concurrent.atomic.AtomicInteger;
public class AsyncResourcesTest
{
@@ -123,62 +121,4 @@ public class AsyncResourcesTest
Assertions.assertThrows(DruidException.class, resource::get);
}
- @Test
- public void testFromCloseableFutureClosesResultExactlyOnceOnClose()
- {
- final SettableFuture<CloseableProbe> future = SettableFuture.create();
- final AsyncResource<CloseableProbe> resource =
AsyncResources.fromFutureCloseable(future);
- final CloseableProbe probe = new CloseableProbe();
-
- future.set(probe);
- Assertions.assertTrue(resource.isReady());
- Assertions.assertSame(probe, resource.get());
- Assertions.assertEquals(0, probe.closeCount.get(), "result must not be
closed while the resource is open");
-
- resource.close();
- Assertions.assertEquals(1, probe.closeCount.get(), "closing the resource
closes the managed result");
- }
-
- @Test
- public void testFromCloseableFutureFailureSurfacesAtGet()
- {
- final SettableFuture<CloseableProbe> future = SettableFuture.create();
- final AsyncResource<CloseableProbe> resource =
AsyncResources.fromFutureCloseable(future);
-
- final RuntimeException boom = new IllegalStateException("boom");
- future.setException(boom);
- Assertions.assertSame(boom,
Assertions.assertThrows(RuntimeException.class, resource::get));
- }
-
- @Test
- public void
testFromCloseableFutureCloseBeforeCompleteStillClosesTheLateResult()
- {
- final SettableFuture<CloseableProbe> future = SettableFuture.create();
- final AsyncResource<CloseableProbe> resource =
AsyncResources.fromFutureCloseable(future);
-
- Assertions.assertFalse(resource.isReady());
- resource.close();
-
- // The future must be left alone: a canceled future silently discards a
value set afterwards, so the callback
- // below would never see the result and nothing would ever close it.
- Assertions.assertFalse(future.isCancelled(), "closing must not cancel the
future");
-
- final CloseableProbe probe = new CloseableProbe();
- Assertions.assertTrue(future.set(probe), "the producer's completion must
still be accepted");
- Assertions.assertEquals(1, probe.closeCount.get(), "a result produced
after close must be closed, not leaked");
- }
-
- /**
- * Closeable that records how many times it was closed.
- */
- private static class CloseableProbe implements Closeable
- {
- private final AtomicInteger closeCount = new AtomicInteger(0);
-
- @Override
- public void close()
- {
- closeCount.incrementAndGet();
- }
- }
}
diff --git
a/server/src/main/java/org/apache/druid/segment/loading/StorageLoadingThreadPool.java
b/server/src/main/java/org/apache/druid/segment/loading/StorageLoadingThreadPool.java
index f2c802ff21d..cc61964e9ae 100644
---
a/server/src/main/java/org/apache/druid/segment/loading/StorageLoadingThreadPool.java
+++
b/server/src/main/java/org/apache/druid/segment/loading/StorageLoadingThreadPool.java
@@ -183,8 +183,9 @@ public class StorageLoadingThreadPool
* exposing the task's (non-null) result. The result is treated as a plain
value with no lifecycle: closing the
* returned resource does not close the result; closing it before completion
cancels the task.
*
- * <p>This is the unmanaged counterpart of {@link
#submitCloseableAsyncResource}; use that when the task's result
- * owns a lifecycle.
+ * <p>The result must <b>not</b> own a lifecycle, since nothing here would
close it. A task producing something
+ * closeable should populate a {@link
org.apache.druid.common.asyncresource.SettableAsyncResource} itself, so that a
+ * result produced after the consumer gave up still gets closed rather than
dropped by the canceled task.
*
* @see AsyncResources#fromFutureUnmanaged
*/
@@ -193,24 +194,6 @@ public class StorageLoadingThreadPool
return
AsyncResources.fromFutureUnmanaged(getExecutorService().submit(task));
}
- /**
- * Submit a task whose result <b>owns a lifecycle</b> and hand back an
{@link AsyncResource} that manages it: closing
- * the returned resource closes the result, and a result produced after that
close is closed rather than leaked.
- *
- * <p>Closing before completion does not cancel the task: it runs to
completion and its result is closed on arrival.
- * Interrupting it would let the task hand a lifecycle-owning result to a
canceled future, which drops it with
- * nothing left to close it.
- *
- * <p>This is the managed counterpart of {@link
#submitUnmanagedAsyncResource}; use that when the task's result is a
- * plain value with no lifecycle.
- *
- * @see AsyncResources#fromFutureCloseable
- */
- public <T extends Closeable> AsyncResource<T>
submitCloseableAsyncResource(Callable<T> task)
- {
- return
AsyncResources.fromFutureCloseable(getExecutorService().submit(task));
- }
-
@LifecycleStop
public void stop()
{
diff --git
a/server/src/main/java/org/apache/druid/segment/loading/external/StorageLocationVirtualStorageManager.java
b/server/src/main/java/org/apache/druid/segment/loading/external/StorageLocationVirtualStorageManager.java
index 48243b607e5..d2594b95b9a 100644
---
a/server/src/main/java/org/apache/druid/segment/loading/external/StorageLocationVirtualStorageManager.java
+++
b/server/src/main/java/org/apache/druid/segment/loading/external/StorageLocationVirtualStorageManager.java
@@ -31,12 +31,14 @@ import org.apache.druid.segment.loading.CacheEntry;
import org.apache.druid.segment.loading.StorageLoadingThreadPool;
import org.apache.druid.segment.loading.StorageLocation;
import org.apache.druid.segment.loading.StorageLocationSelectorStrategy;
+import org.apache.druid.utils.CloseableUtils;
import javax.annotation.Nullable;
import java.io.File;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.LongSupplier;
@@ -239,9 +241,23 @@ public class StorageLocationVirtualStorageManager
implements VirtualStorageManag
}
if (loadingThreadPool.isAvailable()) {
- return loadingThreadPool.submitCloseableAsyncResource(
- () -> reserveAndPopulate(identifier, sizeSupplier, populator)
- );
+ final SettableAsyncResource<CachedFile> resource = new
SettableAsyncResource<>();
+ final Future<?> future =
loadingThreadPool.getExecutorService().submit(() -> {
+ try {
+ final CachedFile populated = reserveAndPopulate(identifier,
sizeSupplier, populator);
+ if (!resource.set(ResourceHolder.fromCloseable(populated))) {
+ CloseableUtils.closeAndSuppressExceptions(
+ populated,
+ e -> log.warn(e, "Failed to release abandoned cache entry for
identifier[%s]", identifier)
+ );
+ }
+ }
+ catch (Throwable t) {
+ resource.setException(t);
+ }
+ });
+ resource.setCanceler(() -> future.cancel(true));
+ return resource;
} else {
final SettableAsyncResource<CachedFile> resource = new
SettableAsyncResource<>();
try {
diff --git
a/server/src/test/java/org/apache/druid/segment/loading/external/StorageLocationVirtualStorageManagerTest.java
b/server/src/test/java/org/apache/druid/segment/loading/external/StorageLocationVirtualStorageManagerTest.java
index e96f853d93c..858772e8bbc 100644
---
a/server/src/test/java/org/apache/druid/segment/loading/external/StorageLocationVirtualStorageManagerTest.java
+++
b/server/src/test/java/org/apache/druid/segment/loading/external/StorageLocationVirtualStorageManagerTest.java
@@ -25,6 +25,7 @@ import org.apache.druid.error.DruidException;
import org.apache.druid.error.DruidExceptionMatcher;
import org.apache.druid.java.util.common.concurrent.Execs;
import
org.apache.druid.segment.loading.LeastBytesUsedStorageLocationSelectorStrategy;
+import org.apache.druid.segment.loading.SegmentLoaderConfig;
import org.apache.druid.segment.loading.StorageLoadingThreadPool;
import org.apache.druid.segment.loading.StorageLocation;
import org.apache.druid.testing.TemporaryFolderExtension;
@@ -46,6 +47,7 @@ import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
public class StorageLocationVirtualStorageManagerTest
@@ -88,12 +90,101 @@ public class StorageLocationVirtualStorageManagerTest
}
}
+ private void withAsyncManager(AsyncManagerConsumer testLogic) throws
Exception
+ {
+ final StorageLoadingThreadPool pool =
StorageLoadingThreadPool.createFromConfig(
+ SegmentLoaderConfig.builder().virtualStorage(true).build()
+ );
+ try {
+ testLogic.accept(
+ new StorageLocationVirtualStorageManager(
+ Collections.singletonList(location),
+ new
LeastBytesUsedStorageLocationSelectorStrategy(Collections.singletonList(location)),
+ pool
+ )
+ );
+ }
+ finally {
+ pool.stop();
+ }
+ }
@FunctionalInterface
private interface ExecutorConsumer
{
void accept(ExecutorService executor) throws Exception;
}
+ @FunctionalInterface
+ private interface AsyncManagerConsumer
+ {
+ void accept(StorageLocationVirtualStorageManager asyncManager) throws
Exception;
+ }
+
+ @Test
+ public void testReserveAndPopulateAsyncPopulatesOnThePool() throws Exception
+ {
+ withAsyncManager(asyncManager -> {
+ final AsyncResource<CachedFile> resource =
asyncManager.reserveAndPopulateAsync(
+ "async-file",
+ () -> 1000L,
+ file -> Files.write(file.toPath(), new byte[]{1, 2, 3})
+ );
+
+ // The resource owns the CachedFile, so it is closed through the
resource rather than directly.
+ final CachedFile cachedFile = resource.await();
+ Assertions.assertTrue(cachedFile.getFile().exists());
+ resource.close();
+ });
+ }
+
+ @Test
+ public void
testReserveAndPopulateAsyncReleasesTheHoldWhenAbandonedMidPopulate() throws
Exception
+ {
+ withAsyncManager(asyncManager -> {
+ final long capacity = location.availableSizeBytes();
+ final long smallSize = capacity / 4;
+ final long bigSize = capacity - 100;
+
+ final CountDownLatch started = new CountDownLatch(1);
+ final AtomicBoolean release = new AtomicBoolean(false);
+
+ final AsyncResource<CachedFile> resource =
asyncManager.reserveAndPopulateAsync(
+ "abandoned",
+ () -> smallSize,
+ file -> {
+ started.countDown();
+ while (!release.get()) {
+ Thread.onSpinWait();
+ }
+ // Model a populate that finishes despite the cancellation:
clearing the interrupt keeps the write below
+ // from aborting, so the task really does produce a CachedFile
after its consumer has gone away.
+ Thread.interrupted();
+ Files.write(file.toPath(), new byte[]{1, 2, 3});
+ }
+ );
+
+ Assertions.assertTrue(started.await(30, TimeUnit.SECONDS), "populate
must be underway before abandoning it");
+ resource.close();
+ release.set(true);
+
+ // The abandoned CachedFile has a reservation hold. If it is leaked the
entry stays pinned, and reserving a
+ // file that needs its space can never succeed; retry until the populate
finishes so this fails only on a leak.
+ final long deadlineNanos = System.nanoTime() +
TimeUnit.SECONDS.toNanos(30);
+ CachedFile big = null;
+ while (big == null && System.nanoTime() < deadlineNanos) {
+ try {
+ big = manager.reserveAndPopulate("big", () -> bigSize, file ->
Files.write(file.toPath(), new byte[]{4}));
+ }
+ catch (DruidException e) {
+ // Still pinned: either the populate is in flight, or its hold
leaked.
+ }
+ }
+
+ Assertions.assertNotNull(big, "the abandoned populate's reservation hold
must be released, not leaked");
+ big.close();
+ });
+ }
+
@Test
public void testGetReturnsNullWhenNotExists()
{
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
