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 45128436a9a fix: SettableAsyncResource fire callbacks if closed before
set (#20074)
45128436a9a is described below
commit 45128436a9a0bc44f8cd60a900cd3d8a58ae02be
Author: Clint Wylie <[email protected]>
AuthorDate: Fri Sep 11 13:03:58 2026 -0700
fix: SettableAsyncResource fire callbacks if closed before set (#20074)
---
.../druid/common/asyncresource/AsyncResource.java | 47 ++++++--
.../AsyncResourceCanceledException.java | 35 ++++++
.../druid/common/asyncresource/AsyncResources.java | 37 ++++--
.../common/asyncresource/RecoverAsyncResource.java | 6 +
.../asyncresource/SettableAsyncResource.java | 43 ++++++-
.../apache/druid/segment/AsyncCursorHolder.java | 5 +-
.../common/asyncresource/AsyncResourcesTest.java | 11 +-
.../asyncresource/CollectAsyncResourceTest.java | 62 ++++++++++
.../asyncresource/RecoverAsyncResourceTest.java | 49 +++++++-
.../asyncresource/SettableAsyncResourceTest.java | 125 +++++++++++++++++++++
.../asyncresource/TransformAsyncResourceTest.java | 26 +++++
.../druid/segment/AsyncCursorHolderTest.java | 19 ++++
.../segment/loading/StorageLoadingThreadPool.java | 7 +-
13 files changed, 438 insertions(+), 34 deletions(-)
diff --git
a/processing/src/main/java/org/apache/druid/common/asyncresource/AsyncResource.java
b/processing/src/main/java/org/apache/druid/common/asyncresource/AsyncResource.java
index 6fd8057ecb5..8c4e60688b0 100644
---
a/processing/src/main/java/org/apache/druid/common/asyncresource/AsyncResource.java
+++
b/processing/src/main/java/org/apache/druid/common/asyncresource/AsyncResource.java
@@ -34,7 +34,9 @@ import java.util.concurrent.TimeoutException;
* be used for resources that do not need cleanup or are not acquired
asynchronously, but it is most useful when
* both are true. The wrapper generally owns the resource lifecycle; see "to
consume a resource" below for details.
*
- * <p>To produce a resource, generally you should create and populate {@link
SettableAsyncResource}.
+ * <p>To produce a resource, generally you should create and populate {@link
SettableAsyncResource}: complete it
+ * exactly once with {@link SettableAsyncResource#set} or {@link
SettableAsyncResource#setException}, and leave
+ * {@link #close()} to the consumer, which owns the resource and may close it
to cancel acquisition early.
*
* <p>To consume a resource, use {@link #addReadyCallback(Runnable)}, {@link
#await()}, or {@link #await(long)}
* to wait for the resource to become ready. Then use {@link #get()} to
retrieve the resource. When you are done
@@ -68,22 +70,33 @@ import java.util.concurrent.TimeoutException;
* cancellation, and in this case, the resource becomes eligible for GC
without completing the future and therefore
* without being closed.
*
- * <p>AsyncResource handles this problem by automatically closing the resource
in
- * {@link SettableAsyncResource#set(ResourceHolder)} when the {@link
SettableAsyncResource} has been canceled.
+ * <p>AsyncResource handles this problem by reporting the race to the producer
instead of dropping the resource: once
+ * this {@link AsyncResource} has been closed, {@link
SettableAsyncResource#set(ResourceHolder)} becomes a no-op and
+ * returns false, which tells the producer that the object it was handing over
is orphaned and that closing it is now
+ * the producer's job. Acquisition can also be canceled on the producer side,
via
+ * {@link SettableAsyncResource#setCanceler(Runnable)}.
*/
public interface AsyncResource<T> extends Closeable
{
/**
- * Whether resource acquisition has completed (successfully or with
failure). To wait for this to become true
- * asynchronously, use {@link #addReadyCallback(Runnable)}. To block until
readiness, use {@link #await()}
- * or {@link #await(long)}.
+ * Whether resource acquisition is no longer in progress, i.e. it succeeded,
failed, was canceled by
+ * {@link #close()}, or was released by {@link
SettableAsyncResource#release()}. Never goes back to false once true;
+ * use {@link #get()} to find out which of those happened. To wait for this
to become true asynchronously, use
+ * {@link #addReadyCallback(Runnable)}. To block until readiness, use {@link
#await()} or {@link #await(long)}.
*/
boolean isReady();
/**
* Register a callback to fire when {@link #isReady()} becomes true (whether
the load succeeded or failed). If the
- * holder is already ready, the callback fires immediately in the calling
thread. Callbacks are not fired if
- * {@link #close()} is called prior to the resource becoming available.
+ * holder is already ready, the callback fires immediately in the calling
thread. Callbacks also fire when
+ * {@link #close()} cancels acquisition before the resource became
available, so that a waiting consumer learns it
+ * was aborted; {@link #get()} then throws {@link
AsyncResourceCanceledException}.
+ *
+ * <p>Firing on close looks redundant, since the same owner both registers
the callbacks and does the closing, but it
+ * lets that owner cancel itself in one call: when a callback completes
something downstream, such as a future
+ * holding a query's result, closing the resource runs the callback, which
handles
+ * {@link AsyncResourceCanceledException} from {@link #get()} and unwinds
the waiter too, with no separate
+ * cancellation step.
*
* <p>Because of the fires-immediately case, the callback can run on the
REGISTERING thread, not just on whatever
* thread completes the resource, so a callback must not do blocking or
expensive work (I/O, deserialization)
@@ -96,8 +109,9 @@ public interface AsyncResource<T> extends Closeable
/**
* Retrieve the underlying object. May be called any number of times, and
the same object will be returned.
*
- * <p>Throws {@link DruidException} if the underlying object is not ready or
if {@link #close()} has been called.
- * Also throws an exception if the resource acquisition failed.
+ * <p>Throws {@link AsyncResourceCanceledException} if {@link #close()}
canceled acquisition before it completed, and
+ * {@link DruidException} if the underlying object is not ready or was
closed after becoming ready. Also throws an
+ * exception if the resource acquisition failed.
*/
T get();
@@ -105,6 +119,9 @@ public interface AsyncResource<T> extends Closeable
* Block until {@link #isReady()} returns true. Does not close the resource
if interrupted; callers must still
* call {@link #close()}.
*
+ * <p>A {@link #close()} that cancels acquisition wakes the waiter, which
then throws
+ * {@link AsyncResourceCanceledException}.
+ *
* <p>Throws {@link DruidException} if {@link #close()} has been called
prior to this method.
*/
default T await() throws InterruptedException
@@ -119,6 +136,9 @@ public interface AsyncResource<T> extends Closeable
* Block until {@link #isReady()} returns true, up to some timeout. Does not
close the resource if interrupted
* or if waiting times out; callers must still call {@link #close()}.
*
+ * <p>A {@link #close()} that cancels acquisition wakes the waiter, which
then throws
+ * {@link AsyncResourceCanceledException}.
+ *
* <p>Throws {@link DruidException} if {@link #close()} has been called
prior to this method.
*/
default T await(long timeoutMillis) throws InterruptedException,
TimeoutException
@@ -133,7 +153,12 @@ public interface AsyncResource<T> extends Closeable
/**
* Closes the resource if it is ready, and has not been released by {@link
SettableAsyncResource#release()}.
- * If acquisition is still in progress, it is canceled if possible.
+ * If acquisition is still in progress, it is canceled if possible, and any
pending
+ * {@link #addReadyCallback(Runnable)} callbacks fire so that waiting
consumers learn acquisition was aborted
+ * instead of waiting for a completion that will never come. {@link
#isReady()} then returns true and
+ * {@link #get()} throws {@link AsyncResourceCanceledException}.
+ *
+ * <p>Only the owner of this resource (the consumer) should call this method.
*
* <p>Despite {@link Closeable} requiring this method to be idempotent, it
is not necessarily
* going to be idempotent. Do not close more than once.
diff --git
a/processing/src/main/java/org/apache/druid/common/asyncresource/AsyncResourceCanceledException.java
b/processing/src/main/java/org/apache/druid/common/asyncresource/AsyncResourceCanceledException.java
new file mode 100644
index 00000000000..05beeab39fb
--- /dev/null
+++
b/processing/src/main/java/org/apache/druid/common/asyncresource/AsyncResourceCanceledException.java
@@ -0,0 +1,35 @@
+/*
+ * 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.common.asyncresource;
+
+import java.util.concurrent.CancellationException;
+
+/**
+ * Thrown by {@link AsyncResource#get()} when {@link AsyncResource#close()}
canceled acquisition before the resource
+ * became available, i.e. the consumer that owned the resource gave up waiting
for it. Extends
+ * {@link CancellationException} so that consumers who only care that
something was canceled need no special handling.
+ */
+public class AsyncResourceCanceledException extends CancellationException
+{
+ public AsyncResourceCanceledException(String message)
+ {
+ super(message);
+ }
+}
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 91d5a7f837b..a7832937106 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
@@ -98,8 +98,15 @@ public class AsyncResources
* 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 (a cancel/close-vs-completion race) is
closed rather than leaked. Closing the
- * returned resource before the future completes cancels the future ({@code
cancel(true)}).
+ * 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.
@@ -107,7 +114,6 @@ public class AsyncResources
public static <T extends Closeable> AsyncResource<T>
fromFutureCloseable(final ListenableFuture<T> future)
{
final SettableAsyncResource<T> retVal = new SettableAsyncResource<>();
- retVal.setCanceler(() -> future.cancel(true));
Futures.addCallback(
future,
new FutureCallback<>()
@@ -115,7 +121,7 @@ public class AsyncResources
@Override
public void onSuccess(T result)
{
- // Lost the race with close()/cancel(): the resource is already
closed, so set() returns false and we own
+ // 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 -> {});
@@ -134,9 +140,11 @@ public class AsyncResources
}
/**
- * Returns an {@link AsyncResource} that collects a list of underlying
resources into a single lifecycle.
- * Calling {@link AsyncResource#close()} on the returned async resource
causes the underlying async resource
- * to be closed.
+ * Returns an {@link AsyncResource} whose value is the result of calling
{@code function} on an underlying resource.
+ *
+ * <p>Once this method returns, the returned {@link AsyncResource} is the
caller's to close, and closing it also
+ * closes {@code sourceResource}, so the caller must not close {@code
sourceResource} itself. If this method throws,
+ * nothing has been taken over and the caller still owns {@code
sourceResource}.
*
* <p>The transformation generally happens eagerly in the thread that
provides the source resource, so it is
* important that it run quickly.
@@ -155,8 +163,10 @@ public class AsyncResources
/**
* Returns an {@link AsyncResource} that collects a list of underlying
resources into a single lifecycle.
- * Calling {@link AsyncResource#close()} on the returned async resource
causes the underlying async resources
- * to also be closed.
+ *
+ * <p>Once this method returns, the returned {@link AsyncResource} is the
caller's to close, and closing it also
+ * closes every resource in {@code asyncResources}, so the caller must not
close them itself. If this method throws,
+ * nothing has been taken over and the caller still owns all of them.
*/
public static <T> AsyncResource<List<T>> collect(final
List<AsyncResource<T>> asyncResources)
{
@@ -171,8 +181,13 @@ public class AsyncResources
* given a chance to substitute a fallback value. Recovery generally happens
eagerly in the thread that provides
* the source resource, so it is important that it run quickly.
*
- * <p>When recovery happens, the {@code sourceResource} is closed
immediately. Otherwise, the {@code sourceResoruce}
- * is closed when the resource returned by this function is closed.
+ * <p>The {@code recoverFn} is not called when the source was canceled by
{@link AsyncResource#close()}: there is
+ * no consumer left to recover for.
+ *
+ * <p>Once this method returns, the returned {@link AsyncResource} is the
caller's to close and the caller must not
+ * close {@code sourceResource} itself: it is closed immediately when
recovery happens, and otherwise when the
+ * returned resource is closed. If this method throws, nothing has been
taken over and the caller still owns
+ * {@code sourceResource}.
*
* <p>The target of {@code function} need not be {@link Closeable}, and even
if it is {@link Closeable}, it
* is not closed (only the source is closed). This transform utility is
meant for transformations that do
diff --git
a/processing/src/main/java/org/apache/druid/common/asyncresource/RecoverAsyncResource.java
b/processing/src/main/java/org/apache/druid/common/asyncresource/RecoverAsyncResource.java
index a83a70fd74f..5e64dc6f5d1 100644
---
a/processing/src/main/java/org/apache/druid/common/asyncresource/RecoverAsyncResource.java
+++
b/processing/src/main/java/org/apache/druid/common/asyncresource/RecoverAsyncResource.java
@@ -93,6 +93,12 @@ public class RecoverAsyncResource<T> implements
AsyncResource<T>
try {
value = sourceResource.get();
}
+ catch (AsyncResourceCanceledException e) {
+ // The source was closed before it became available, so this callback is
firing from its close() and there is no
+ // consumer left to recover for.
+ targetResource.setException(e);
+ return;
+ }
catch (Throwable e) {
final T recovered;
try {
diff --git
a/processing/src/main/java/org/apache/druid/common/asyncresource/SettableAsyncResource.java
b/processing/src/main/java/org/apache/druid/common/asyncresource/SettableAsyncResource.java
index 9894d16a311..050568e2763 100644
---
a/processing/src/main/java/org/apache/druid/common/asyncresource/SettableAsyncResource.java
+++
b/processing/src/main/java/org/apache/druid/common/asyncresource/SettableAsyncResource.java
@@ -81,6 +81,13 @@ public class SettableAsyncResource<T> implements
AsyncResource<T>
@GuardedBy("this")
private Throwable error = null;
+ /**
+ * Whether {@link #close()} happened before the resource became available,
i.e. acquisition was canceled. Such a
+ * resource is complete: {@link #isReady()} is true and {@link #get()}
throws {@link AsyncResourceCanceledException}.
+ */
+ @GuardedBy("this")
+ private boolean canceled = false;
+
@GuardedBy("this")
private State state = State.NEW;
@@ -126,7 +133,8 @@ public class SettableAsyncResource<T> implements
AsyncResource<T>
* {@link #addReadyCallback(Runnable)}. Once this method returns true,
{@link #close()} will no longer call
* the canceler from {@link #setCanceler(Runnable)}.
*
- * <p>If this method returns false, the producer is responsible for closing
the resource itself.
+ * <p>If this method returns false, the resource was already closed: the
producer is responsible for closing the
+ * resource itself. The pending callbacks already fired when {@link
#close()} canceled acquisition.
*
* <p>Throws {@link DruidException} if this resource was already completed
from a prior call to this method or
* {@link #setException}).
@@ -169,8 +177,11 @@ public class SettableAsyncResource<T> implements
AsyncResource<T>
* {@link #addReadyCallback(Runnable)}. Afterwards, {@link #close()} will no
longer call the canceler from
* {@link #setCanceler(Runnable)}.
*
+ * <p>If the resource was already closed, the error is logged at debug and
otherwise dropped: the consumer already
+ * learned that acquisition was canceled when {@link #close()} fired the
pending callbacks.
+ *
* <p>Throws {@link DruidException} if this resource was already completed
from a prior call to this method or
- * {@link #set}).
+ * {@link #set}.
*/
public void setException(Throwable t)
{
@@ -180,7 +191,9 @@ public class SettableAsyncResource<T> implements
AsyncResource<T>
@Override
public synchronized boolean isReady()
{
- return state == State.READY;
+ // anything but NEW means acquisition is over, whether it succeeded,
failed, was canceled, or was released; get()
+ // reports which. Once true this stays true, so a consumer waiting on it
can never be sent back to waiting.
+ return state != State.NEW;
}
@Override
@@ -199,7 +212,12 @@ public class SettableAsyncResource<T> implements
AsyncResource<T>
}
}
case RELEASED -> throw DruidException.defensive("Resource has been
released");
- case CLOSED -> throw DruidException.defensive("Closed");
+ case CLOSED -> {
+ if (canceled) {
+ throw new AsyncResourceCanceledException("Resource acquisition was
canceled by close()");
+ }
+ throw DruidException.defensive("Closed");
+ }
};
}
@@ -215,8 +233,8 @@ public class SettableAsyncResource<T> implements
AsyncResource<T>
* {@link AsyncResources#collect}, {@link AsyncResources#transform}, etc.
These combinators will fail to properly
* encapsulate resource lifecycle if resources have been released.
*
- * <p>Throws {@link DruidException} if the holder is not yet ready, has
already been released, or if
- * {@link #close()} has been called.
+ * <p>Throws {@link DruidException} if the holder is not yet ready, has
already been released, or was closed after
+ * becoming ready. Throws {@link AsyncResourceCanceledException} if {@link
#close()} canceled acquisition first.
*/
protected synchronized T release()
{
@@ -254,6 +272,7 @@ public class SettableAsyncResource<T> implements
AsyncResource<T>
public void close()
{
final Closeable deferredCloseable;
+ final List<Runnable> callbacksToFire;
synchronized (this) {
deferredCloseable = switch (state) {
@@ -263,6 +282,9 @@ public class SettableAsyncResource<T> implements
AsyncResource<T>
default -> throw DruidException.defensive("Already closed");
};
+ canceled = state == State.NEW;
+ callbacksToFire = drainCallbacks();
+
// Clear result and canceler to allow GC.
result = null;
canceler = null;
@@ -273,6 +295,8 @@ public class SettableAsyncResource<T> implements
AsyncResource<T>
deferredCloseable,
e -> LOG.warn(e, "Failed to call cleaner of class[%s]",
deferredCloseable.getClass())
);
+
+ fireCallbacks(callbacksToFire);
}
@GuardedBy("this")
@@ -308,6 +332,13 @@ public class SettableAsyncResource<T> implements
AsyncResource<T>
canceler = null;
callbacksToFire = drainCallbacks();
}
+
+ if (!didSet && value.isError()) {
+ // Nothing will ever surface this error: the callbacks already fired at
close() and get() reports the
+ // cancellation, so debug log it rather than let a failure that lost a
race with close() vanish.
+ LOG.debug(value.error(), "Resource failed after close().");
+ }
+
fireCallbacks(callbacksToFire);
return didSet;
}
diff --git
a/processing/src/main/java/org/apache/druid/segment/AsyncCursorHolder.java
b/processing/src/main/java/org/apache/druid/segment/AsyncCursorHolder.java
index d980022d325..fa238954e6f 100644
--- a/processing/src/main/java/org/apache/druid/segment/AsyncCursorHolder.java
+++ b/processing/src/main/java/org/apache/druid/segment/AsyncCursorHolder.java
@@ -35,7 +35,10 @@ import javax.annotation.Nullable;
* <h3>Consumer protocol</h3>
* Consumers wait for {@link #isReady()} via {@link #addReadyCallback}, and
{@link #release()} to transfer ownership of
* the {@link CursorHolder} (or throw the producer exception). Calling {@link
#release()} before {@link #isReady()}
- * returns {@code true}, multiple times, or after this holder has been closed
will throw a {@link DruidException}.
+ * returns {@code true}, multiple times, or after this holder has been closed
will throw a {@link DruidException}. If
+ * {@link #close()} canceled the load before it completed, {@link #isReady()}
becomes true and {@link #release()}
+ * throws {@link
org.apache.druid.common.asyncresource.AsyncResourceCanceledException} instead,
so a waiting consumer
+ * always finds out.
* <p>
* For example (using {@link ReturnOrAwait} to show intended yield-then-resume
usage pattern):
* <pre>{@code
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 dc25f7479c3..4e036179308 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
@@ -151,14 +151,21 @@ public class AsyncResourcesTest
}
@Test
- public void testFromCloseableFutureCloseBeforeCompleteCancelsFuture()
+ public void
testFromCloseableFutureCloseBeforeCompleteStillClosesTheLateResult()
{
final SettableFuture<CloseableProbe> future = SettableFuture.create();
final AsyncResource<CloseableProbe> resource =
AsyncResources.fromFutureCloseable(future);
Assertions.assertFalse(resource.isReady());
resource.close();
- Assertions.assertTrue(future.isCancelled(), "closing before completion
cancels the backing future");
+
+ // 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");
}
/**
diff --git
a/processing/src/test/java/org/apache/druid/common/asyncresource/CollectAsyncResourceTest.java
b/processing/src/test/java/org/apache/druid/common/asyncresource/CollectAsyncResourceTest.java
index 40ce3e9b4bc..e71e19c8054 100644
---
a/processing/src/test/java/org/apache/druid/common/asyncresource/CollectAsyncResourceTest.java
+++
b/processing/src/test/java/org/apache/druid/common/asyncresource/CollectAsyncResourceTest.java
@@ -19,6 +19,7 @@
package org.apache.druid.common.asyncresource;
+import org.apache.druid.error.DruidException;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -128,4 +129,65 @@ public class CollectAsyncResourceTest
collected.close();
Assertions.assertEquals(1, aCancel.get(), "closing the collect must cancel
pending sources");
}
+
+ @Test
+ public void testCloseBeforeReadyWakesConsumerAndReportsCancellation()
+ {
+ final AtomicInteger fired = new AtomicInteger();
+ final SettableAsyncResource<String> a = new SettableAsyncResource<>();
+ final SettableAsyncResource<String> b = new SettableAsyncResource<>();
+
+ final AsyncResource<List<String>> collected =
AsyncResources.collect(List.of(a, b));
+ collected.addReadyCallback(fired::incrementAndGet);
+
+ // Closing must reach a consumer waiting on the collected resource, not
just cancel the sources. Each source's
+ // close also drives this class's own onOneSourceReady, which must not
complete or re-fire anything.
+ collected.close();
+
+ Assertions.assertEquals(1, fired.get(), "closing must wake a waiting
consumer exactly once");
+ Assertions.assertTrue(collected.isReady());
+ Assertions.assertThrows(AsyncResourceCanceledException.class,
collected::get);
+ }
+
+ @Test
+ public void testCloseWithAMixOfReadyAndPendingSourcesReportsCancellation()
+ {
+ final AtomicInteger fired = new AtomicInteger();
+ final AtomicInteger aClose = new AtomicInteger();
+ final SettableAsyncResource<String> a = new SettableAsyncResource<>();
+ final SettableAsyncResource<String> b = new SettableAsyncResource<>();
+ a.set("a", aClose::incrementAndGet);
+
+ final AsyncResource<List<String>> collected =
AsyncResources.collect(List.of(a, b));
+ collected.addReadyCallback(fired::incrementAndGet);
+ Assertions.assertFalse(collected.isReady(), "one source is still pending");
+
+ // The already-ready source counted toward readiness, so closing the
pending one brings the internal count to the
+ // source count and runs the collect body against sources this close just
tore down. That must stay harmless.
+ collected.close();
+
+ Assertions.assertEquals(1, fired.get());
+ Assertions.assertEquals(1, aClose.get(), "the ready source's value must
still be closed");
+ Assertions.assertThrows(AsyncResourceCanceledException.class,
collected::get);
+ }
+
+ @Test
+ public void testThrowingConstructionLeavesSourcesToTheCaller()
+ {
+ final AtomicInteger aCancel = new AtomicInteger();
+ final SettableAsyncResource<String> a = new SettableAsyncResource<>();
+ a.setCanceler(aCancel::incrementAndGet);
+
+ // A closed source rejects addReadyCallback, so collect throws partway
through registering. Nothing has been taken
+ // over at that point, so the caller still owns every input: collect must
not have canceled or closed any of them.
+ final SettableAsyncResource<String> closed = new SettableAsyncResource<>();
+ closed.close();
+
+ Assertions.assertThrows(DruidException.class, () ->
AsyncResources.collect(List.of(a, closed)));
+ Assertions.assertEquals(0, aCancel.get(), "a failed collect must leave its
inputs alone");
+
+ // Still the caller's to close, and still usable.
+ a.close();
+ Assertions.assertEquals(1, aCancel.get());
+ }
}
diff --git
a/processing/src/test/java/org/apache/druid/common/asyncresource/RecoverAsyncResourceTest.java
b/processing/src/test/java/org/apache/druid/common/asyncresource/RecoverAsyncResourceTest.java
index 9a3c695de61..8effaec0ea9 100644
---
a/processing/src/test/java/org/apache/druid/common/asyncresource/RecoverAsyncResourceTest.java
+++
b/processing/src/test/java/org/apache/druid/common/asyncresource/RecoverAsyncResourceTest.java
@@ -22,6 +22,7 @@ package org.apache.druid.common.asyncresource;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.util.concurrent.CancellationException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
@@ -141,12 +142,58 @@ public class RecoverAsyncResourceTest
// Close the wrapper while the source is still pending.
recovering.close();
Assertions.assertEquals(1, source.closeCount.get());
- Assertions.assertFalse(recovering.isReady());
+ // Canceled acquisition completes the resource, so waiting consumers stop
waiting.
+ Assertions.assertTrue(recovering.isReady());
// Completing the source after close is a no-op and must not throw.
Assertions.assertFalse(source.delegate.set("late", null));
}
+ @Test
+ public void testCloseCancellationIsNotRecoveredFrom()
+ {
+ final TrackingAsyncResource<String> source = new TrackingAsyncResource<>();
+ final AtomicInteger recoveryCalls = new AtomicInteger();
+
+ final AsyncResource<String> recovering = AsyncResources.recover(
+ source,
+ e -> {
+ recoveryCalls.incrementAndGet();
+ return "fallback";
+ }
+ );
+
+ // Closing fires the source's ready callback, which lands in this class's
failure path. Recovering there would run
+ // the fallback for a consumer that has already gone away, so cancellation
must pass straight through.
+ recovering.close();
+
+ Assertions.assertEquals(0, recoveryCalls.get(), "recovery must not be
called for a canceled source");
+ Assertions.assertThrows(AsyncResourceCanceledException.class,
recovering::get);
+ }
+
+ @Test
+ public void testProducerCancellationIsStillRecoveredFrom()
+ {
+ final TrackingAsyncResource<String> source = new TrackingAsyncResource<>();
+ final AtomicInteger recoveryCalls = new AtomicInteger();
+
+ final AsyncResource<String> recovering = AsyncResources.recover(
+ source,
+ e -> {
+ recoveryCalls.incrementAndGet();
+ return "fallback";
+ }
+ );
+
+ // A load canceled underneath a consumer that is still waiting (an
executor shutdown, say) arrives as a plain
+ // CancellationException rather than AsyncResourceCanceledException. There
is someone left to serve, so unlike a
+ // close-initiated cancellation this is a failure the fallback should
cover.
+ source.delegate.setException(new CancellationException("canceled
underneath a waiting consumer"));
+
+ Assertions.assertEquals(1, recoveryCalls.get(), "recovery must run when
the consumer is still waiting");
+ Assertions.assertEquals("fallback", recovering.get());
+ }
+
/**
* An {@link AsyncResource} that delegates to a {@link
SettableAsyncResource} and counts {@link #close()} calls,
* so tests can verify how the source resource's lifecycle is managed.
diff --git
a/processing/src/test/java/org/apache/druid/common/asyncresource/SettableAsyncResourceTest.java
b/processing/src/test/java/org/apache/druid/common/asyncresource/SettableAsyncResourceTest.java
index 43def411a36..6712ee17096 100644
---
a/processing/src/test/java/org/apache/druid/common/asyncresource/SettableAsyncResourceTest.java
+++
b/processing/src/test/java/org/apache/druid/common/asyncresource/SettableAsyncResourceTest.java
@@ -25,8 +25,12 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
public class SettableAsyncResourceTest
{
@@ -217,6 +221,127 @@ public class SettableAsyncResourceTest
Assertions.assertThrows(DruidException.class, resource::close);
}
+ @Test
+ public void testSetExceptionPassesTheCancellationTypeThrough()
+ {
+ final SettableAsyncResource<String> resource = new
SettableAsyncResource<>();
+ final AsyncResourceCanceledException canceled = new
AsyncResourceCanceledException("canceled upstream");
+
+ // Wrappers forward a closed source's cancellation to their own target
verbatim, so this type is accepted like any
+ // other error rather than being reserved for close(). It then means
"something upstream was closed before it was
+ // ready", which is still what tells RecoverAsyncResource apart from a
load canceled under a waiting consumer.
+ resource.setException(canceled);
+
+ Assertions.assertTrue(resource.isReady());
+ Assertions.assertSame(canceled,
Assertions.assertThrows(AsyncResourceCanceledException.class, resource::get));
+ }
+
+ @Test
+ public void testSetExceptionWithTheCancellationTypeIsDroppedAfterClose()
+ {
+ final SettableAsyncResource<String> resource = new
SettableAsyncResource<>();
+ resource.close();
+
+ // Wrappers propagate a closed source's cancellation to their own
already-closed target on ordinary cancel paths,
+ // where it is dropped like any other late error.
+ resource.setException(new AsyncResourceCanceledException("propagated from
a closed source"));
+ Assertions.assertThrows(AsyncResourceCanceledException.class,
resource::get);
+ }
+
+ @Test
+ public void testCloseBeforeReadyCompletesAsCanceled()
+ {
+ final SettableAsyncResource<String> resource = new
SettableAsyncResource<>();
+ resource.close();
+
+ // Canceled acquisition is a completion: consumers gate their waiting on
isReady(), and get() says what happened.
+ Assertions.assertTrue(resource.isReady());
+ Assertions.assertThrows(AsyncResourceCanceledException.class,
resource::get);
+ }
+
+ @Test
+ public void testCloseAfterReadyIsNotReportedAsCancellation()
+ {
+ final SettableAsyncResource<String> resource = new
SettableAsyncResource<>();
+ resource.set("value", null);
+ resource.close();
+
+ // Using a resource you closed after it was ready is a coding error, not a
cancellation.
+ Assertions.assertThrows(DruidException.class, resource::get);
+ }
+
+ @Test
+ public void testReadyCallbackFiredOnceWhenSetLosesRaceWithClose()
+ {
+ // close() before the resource is available fires the pending callbacks,
so a waiting consumer learns acquisition
+ // was aborted. A producer's late set() that loses the race must not fire
them a second time: close() drained them.
+ final AtomicInteger fired = new AtomicInteger();
+ final SettableAsyncResource<String> resource = new
SettableAsyncResource<>();
+ resource.addReadyCallback(fired::incrementAndGet);
+
+ resource.close();
+ Assertions.assertEquals(1, fired.get());
+
+ Assertions.assertFalse(resource.set("value", null));
+ Assertions.assertEquals(1, fired.get());
+ }
+
+ @Test
+ public void testReadyCallbackFiredOnceWhenSetExceptionLosesRaceWithClose()
+ {
+ final AtomicInteger fired = new AtomicInteger();
+ final SettableAsyncResource<String> resource = new
SettableAsyncResource<>();
+ resource.addReadyCallback(fired::incrementAndGet);
+
+ resource.close();
+ Assertions.assertEquals(1, fired.get());
+
+ // setException on a CLOSED resource is dropped (logged at debug) and must
not re-fire the callback.
+ resource.setException(new IllegalStateException("late failure"));
+ Assertions.assertEquals(1, fired.get());
+ }
+
+ @Test
+ public void testCloseFiresCallbacksAfterRunningTheCanceler()
+ {
+ // Ordering matters: a woken consumer must observe a fully torn-down
resource, not one still mid-abort.
+ final List<String> order = new ArrayList<>();
+ final SettableAsyncResource<String> resource = new
SettableAsyncResource<>();
+ resource.setCanceler(() -> order.add("canceler"));
+ resource.addReadyCallback(() -> order.add("callback"));
+
+ resource.close();
+ Assertions.assertEquals(List.of("canceler", "callback"), order);
+ }
+
+ @Test
+ public void testAwaitIsWokenByCloseAndThrowsCancellation() throws
InterruptedException
+ {
+ final SettableAsyncResource<String> resource = new
SettableAsyncResource<>();
+ final AtomicReference<Throwable> outcome = new AtomicReference<>();
+
+ // A bounded await, so a regression that stops waking the waiter fails
with a TimeoutException instead of hanging.
+ final Thread waiter = new Thread(() -> {
+ try {
+ resource.await(30_000);
+ }
+ catch (Throwable t) {
+ outcome.set(t);
+ }
+ });
+ waiter.start();
+
+ // Wait until the waiter is parked on await()'s latch, so close() cannot
land before its callback is registered.
+ final long deadlineNanos = System.nanoTime() +
TimeUnit.SECONDS.toNanos(30);
+ while (waiter.getState() != Thread.State.TIMED_WAITING &&
System.nanoTime() < deadlineNanos) {
+ Thread.onSpinWait();
+ }
+
+ resource.close();
+ waiter.join();
+ Assertions.assertInstanceOf(AsyncResourceCanceledException.class,
outcome.get());
+ }
+
@Test
public void testAwaitReturnsValueWhenReady() throws InterruptedException
{
diff --git
a/processing/src/test/java/org/apache/druid/common/asyncresource/TransformAsyncResourceTest.java
b/processing/src/test/java/org/apache/druid/common/asyncresource/TransformAsyncResourceTest.java
index 2a066eef7fc..352714bb391 100644
---
a/processing/src/test/java/org/apache/druid/common/asyncresource/TransformAsyncResourceTest.java
+++
b/processing/src/test/java/org/apache/druid/common/asyncresource/TransformAsyncResourceTest.java
@@ -153,4 +153,30 @@ public class TransformAsyncResourceTest
transformed.close();
Assertions.assertEquals(1, sourceCancel.get());
}
+
+ @Test
+ public void testCloseBeforeReadyWakesConsumerAndReportsCancellation()
+ {
+ final AtomicInteger functionCalls = new AtomicInteger();
+ final AtomicInteger fired = new AtomicInteger();
+ final SettableAsyncResource<Integer> source = new
SettableAsyncResource<>();
+
+ final AsyncResource<String> transformed = AsyncResources.transform(
+ source,
+ i -> {
+ functionCalls.incrementAndGet();
+ return "v" + i;
+ }
+ );
+ transformed.addReadyCallback(fired::incrementAndGet);
+
+ // Closing must reach a consumer waiting on the transformed resource, not
just cancel the source. Closing the
+ // source also drives this class's own onSourceReady, which must not fire
the consumer's callback a second time.
+ transformed.close();
+
+ Assertions.assertEquals(1, fired.get(), "closing must wake a waiting
consumer exactly once");
+ Assertions.assertTrue(transformed.isReady());
+ Assertions.assertThrows(AsyncResourceCanceledException.class,
transformed::get);
+ Assertions.assertEquals(0, functionCalls.get(), "the function must not run
for a canceled source");
+ }
}
diff --git
a/processing/src/test/java/org/apache/druid/segment/AsyncCursorHolderTest.java
b/processing/src/test/java/org/apache/druid/segment/AsyncCursorHolderTest.java
index 409f6b6a16b..c001fcc85ca 100644
---
a/processing/src/test/java/org/apache/druid/segment/AsyncCursorHolderTest.java
+++
b/processing/src/test/java/org/apache/druid/segment/AsyncCursorHolderTest.java
@@ -27,6 +27,25 @@ import java.util.concurrent.atomic.AtomicInteger;
class AsyncCursorHolderTest
{
+ @Test
+ void testIsReadyStaysTrueThroughReleaseAndClose()
+ {
+ final CountingCursorHolder holder = new CountingCursorHolder();
+ final AsyncCursorHolder asyncHolder = new AsyncCursorHolder(null);
+ Assertions.assertFalse(asyncHolder.isReady());
+
+ Assertions.assertTrue(asyncHolder.set(holder));
+ Assertions.assertTrue(asyncHolder.isReady());
+
+ // Releasing transfers ownership, and closing afterwards is a no-op, but
acquisition is over either way: readiness
+ // must not flip back to false, or a consumer that gates its waiting on it
could be sent back to waiting.
+ asyncHolder.release();
+ Assertions.assertTrue(asyncHolder.isReady());
+
+ asyncHolder.close();
+ Assertions.assertTrue(asyncHolder.isReady());
+ }
+
@Test
void testCloseAfterReleaseDoesNotDoubleCloseHolder()
{
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 866e5eeeb82..f2c802ff21d 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
@@ -195,8 +195,11 @@ public class StorageLoadingThreadPool
/**
* 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, a result produced after a
cancel/close race is closed rather than leaked,
- * and closing it before completion cancels the task.
+ * 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.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
