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 c206beeb7f5 fix: FrameProcessorExecutor cancellation fixes (#20460)
c206beeb7f5 is described below
commit c206beeb7f5fa0bb07577e83b6f80aa5b1856c4d
Author: Clint Wylie <[email protected]>
AuthorDate: Mon Oct 12 21:26:30 2026 -0700
fix: FrameProcessorExecutor cancellation fixes (#20460)
---
.../frame/processor/FrameProcessorExecutor.java | 61 ++++++++++++++++++++--
.../processor/FrameProcessorExecutorTest.java | 47 ++++++++++++++++-
2 files changed, 102 insertions(+), 6 deletions(-)
diff --git
a/processing/src/main/java/org/apache/druid/frame/processor/FrameProcessorExecutor.java
b/processing/src/main/java/org/apache/druid/frame/processor/FrameProcessorExecutor.java
index 530232d4e07..c7ad7a25518 100644
---
a/processing/src/main/java/org/apache/druid/frame/processor/FrameProcessorExecutor.java
+++
b/processing/src/main/java/org/apache/druid/frame/processor/FrameProcessorExecutor.java
@@ -105,6 +105,10 @@ public class FrameProcessorExecutor
* If "cancellationId" is provided, it must have previously been registered
with {@link #registerCancellationId}.
* Then, it can be used with the {@link #cancel(String)} method to cancel
all processors with that
* same cancellationId.
+ *
+ * This method always cleans the processor up, whether or not it runs
(callers must never clean it up themselves).
+ * The processor is not run at all if "cancellationId" has already been
canceled, in which case the returned future
+ * comes back already canceled. With a null cancellationId the processor
always runs.
*/
public <T> ListenableFuture<T> runFully(final FrameProcessor<T> processor,
@Nullable final String cancellationId)
{
@@ -113,7 +117,8 @@ public class FrameProcessorExecutor
final SettableFuture<T> finished =
registerCancelableFuture(SettableFuture.create(), true, cancellationId);
if (finished.isDone()) {
- // Possibly due to starting life out being canceled.
+ // Canceled either before or just after registerCancelableFuture ran;
clean up because nothing else can
+ cancel(Collections.singleton(processor));
return finished;
}
@@ -451,8 +456,40 @@ public class FrameProcessorExecutor
);
logProcessorStatusString(processor, finished.isDone(), null, null, null);
- registerCancelableProcessor(processor, cancellationId);
- exec.execute(runnable);
+
+ if (!registerCancelableProcessor(processor, cancellationId)) {
+ // cancel(cancellationId) ran after "finished" was registered, so it
already cleaned cancelableProcessors
+ // without seeing this one; clean up because nothing else can, and
cancel "finished" because that cancel
+ // resolves return futures last (after waiting on the processors it did
see)
+ cancel(Collections.singleton(processor));
+ finished.cancel(true);
+ return finished;
+ }
+
+ try {
+ exec.execute(runnable);
+ }
+ catch (Throwable e) {
+ // registered, but never going to run; clean up only if we took the
processor out of cancelableProcessors
+ // ourselves (otherwise a concurrent cancel is doing the cleanup)
+ final boolean doCleanup;
+
+ if (cancellationId != null) {
+ synchronized (lock) {
+ doCleanup = cancelableProcessors.remove(cancellationId, processor);
+ }
+ } else {
+ doCleanup = true;
+ }
+
+ if (doCleanup) {
+ cancel(Collections.singleton(processor));
+ }
+
+ finished.setException(e);
+ throw e;
+ }
+
return finished;
}
@@ -488,6 +525,9 @@ public class FrameProcessorExecutor
* Registers a cancellationId, so it can be provided to {@link #runFully} or
{@link #runAllFully}. To avoid the
* set of active cancellationIds growing without bound, callers must also
call {@link #cancel(String)} on the
* same cancellationId when done using it.
+ *
+ * A cancellationId must also not be registered again once canceled. Nothing
here checks for it, and a run still in
+ * flight under the old registration would silently attach itself to the new
one.
*/
public void registerCancellationId(final String cancellationId)
{
@@ -666,13 +706,26 @@ public class FrameProcessorExecutor
}
}
- private <T> void registerCancelableProcessor(final FrameProcessor<T>
processor, @Nullable final String cancellationId)
+ /**
+ * Register a processor so {@link #cancel(String)} can find it.
+ *
+ * @return true if registered, or if there is no cancellationId to register
under. False if {@code cancellationId}
+ * is already canceled, in which case the caller must not run the processor
and owns cleaning it up.
+ */
+ private <T> boolean registerCancelableProcessor(
+ final FrameProcessor<T> processor,
+ @Nullable final String cancellationId
+ )
{
if (cancellationId != null) {
synchronized (lock) {
+ if (!activeCancellationIds.contains(cancellationId)) {
+ return false;
+ }
cancelableProcessors.put(cancellationId, processor);
}
}
+ return true;
}
/**
diff --git
a/processing/src/test/java/org/apache/druid/frame/processor/FrameProcessorExecutorTest.java
b/processing/src/test/java/org/apache/druid/frame/processor/FrameProcessorExecutorTest.java
index b728c2b7469..5c6a25943ef 100644
---
a/processing/src/test/java/org/apache/druid/frame/processor/FrameProcessorExecutorTest.java
+++
b/processing/src/test/java/org/apache/druid/frame/processor/FrameProcessorExecutorTest.java
@@ -78,7 +78,10 @@ import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CancellationException;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.stream.Collectors;
@@ -391,6 +394,46 @@ public class FrameProcessorExecutorTest
exec.cancel("nonexistent");
}
+ @Test
+ @Timeout(value = 30_000L, unit = TimeUnit.MILLISECONDS)
+ public void
test_runFully_cancelRacingRegistration_leavesNoOrphanProcessor() throws
Exception
+ {
+ final ExecutorService launcher =
Execs.singleThreaded("cancel-race-launcher-%s");
+
+ try {
+ for (int i = 0; i < 300; i++) {
+ final String cancellationId = "race-" + i;
+ final SleepyFrameProcessor processor = new SleepyFrameProcessor();
+ exec.registerCancellationId(cancellationId);
+
+ final CountDownLatch ready = new CountDownLatch(1);
+ final Future<ListenableFuture<Long>> launched = launcher.submit(()
-> {
+ ready.countDown();
+ return exec.runFully(processor, cancellationId);
+ });
+
+ // Release the launcher and cancel from this thread, so the cancel
lands somewhere inside runFully.
+ ready.await();
+ exec.cancel(cancellationId);
+
+ final ListenableFuture<Long> future = launched.get(30,
TimeUnit.SECONDS);
+
+ // Asserted every iteration rather than once at the end, so a leak
is reported against the iteration that
+ // caused it
+ Assertions.assertEquals(
+ 0,
+ exec.cancelableProcessorCount(),
+ StringUtils.format("orphaned processor on iteration[%d]", i)
+ );
+ Assertions.assertTrue(future.isDone(), StringUtils.format("future
resolved on iteration[%d]", i));
+ Assertions.assertTrue(processor.didCleanup(),
StringUtils.format("processor cleaned on iteration[%d]", i));
+ }
+ }
+ finally {
+ launcher.shutdownNow();
+ }
+ }
+
@Test
public void test_runFully_nonexistentCancellationId()
{
@@ -400,11 +443,11 @@ public class FrameProcessorExecutorTest
// Don't registerCancellationId(cancellationId).
final ListenableFuture<Long> future = exec.runFully(processor,
cancellationId);
- // Future should be immediately canceled, without running the processor.
+ // Future should be immediately canceled, without running the processor,
but still cleaned up
Assertions.assertTrue(future.isDone());
Assertions.assertTrue(future.isCancelled());
Assertions.assertFalse(processor.didGetInterrupt());
- Assertions.assertFalse(processor.didCleanup());
+ Assertions.assertTrue(processor.didCleanup());
}
@Test
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
