This is an automated email from the ASF dual-hosted git repository.
gianm 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 66f294f2958 fix: Missed cleanup in runAllFully on certain
cancellations. (#20322)
66f294f2958 is described below
commit 66f294f29589b92464f04c9bfdbd2d6c8f93129b
Author: Gian Merlino <[email protected]>
AuthorDate: Tue Oct 20 15:29:26 2026 -0700
fix: Missed cleanup in runAllFully on certain cancellations. (#20322)
Previously, with no processors outstanding, "finished" was never set,
so the processor manager was never closed and its resources could leak.
---
.../druid/frame/processor/RunAllFullyWidget.java | 2 +
.../frame/processor/RunAllFullyWidgetTest.java | 49 ++++++++++++++++++++++
2 files changed, 51 insertions(+)
diff --git
a/processing/src/main/java/org/apache/druid/frame/processor/RunAllFullyWidget.java
b/processing/src/main/java/org/apache/druid/frame/processor/RunAllFullyWidget.java
index 4b096c01c31..ba465cd8b33 100644
---
a/processing/src/main/java/org/apache/druid/frame/processor/RunAllFullyWidget.java
+++
b/processing/src/main/java/org/apache/druid/frame/processor/RunAllFullyWidget.java
@@ -41,6 +41,7 @@ import java.util.IdentityHashMap;
import java.util.Optional;
import java.util.Queue;
import java.util.Set;
+import java.util.concurrent.CancellationException;
import java.util.concurrent.atomic.AtomicReference;
/**
@@ -160,6 +161,7 @@ public class RunAllFullyWidget<T, ResultType>
// Cleanup if there were no outstanding processors. (If
there were some outstanding ones, future
// cancellation above would trigger cleanup.)
+ finished.compareAndSet(null, Either.error(new
CancellationException("Canceled")));
cleanupIfNoMoreProcessors();
}
}
diff --git
a/processing/src/test/java/org/apache/druid/frame/processor/RunAllFullyWidgetTest.java
b/processing/src/test/java/org/apache/druid/frame/processor/RunAllFullyWidgetTest.java
index 5a2ac401b31..b958608183a 100644
---
a/processing/src/test/java/org/apache/druid/frame/processor/RunAllFullyWidgetTest.java
+++
b/processing/src/test/java/org/apache/druid/frame/processor/RunAllFullyWidgetTest.java
@@ -467,6 +467,29 @@ public class RunAllFullyWidgetTest extends
FrameProcessorExecutorTest.BaseFrameP
Assertions.assertEquals(0, exec.cancelableProcessorCount());
}
+ @Test
+ public void test_runAllFully_cancelWhileWaitingForFirstProcessor()
+ {
+ final String cancellationId = "xyzzy";
+ exec.registerCancellationId(cancellationId);
+
+ // This manager never returns a processor, so all runnables park with no
processors outstanding.
+ final ListenableFuture<Long> future = exec.runAllFully(
+ possiblyDelay(ensureClose(new NeverResolvingProcessorManager())),
+ maxOutstandingProcessors,
+ bouncer,
+ cancellationId
+ );
+
+ exec.cancel(cancellationId);
+
+ Assertions.assertTrue(future.isCancelled());
+
+ // The processor manager must be closed even though no processor was ever
outstanding. (tearDown checks this
+ // too, and additionally checks that it happens exactly once.)
+ Assertions.assertEquals(1, closed.get(), "Processor manager closed");
+ }
+
/**
* Wrap in {@link DelayedProcessorManager} if {@link #delayed} is set.
*/
@@ -644,6 +667,32 @@ public class RunAllFullyWidgetTest extends
FrameProcessorExecutorTest.BaseFrameP
}
}
+ /**
+ * Processor manager whose {@link #next()} future never resolves, so callers
park with no processors outstanding.
+ */
+ private static class NeverResolvingProcessorManager implements
ProcessorManager<Object, Long>
+ {
+ private final SettableFuture<Optional<ProcessorAndCallback<Object>>>
future = SettableFuture.create();
+
+ @Override
+ public ListenableFuture<Optional<ProcessorAndCallback<Object>>> next()
+ {
+ return future;
+ }
+
+ @Override
+ public Long result()
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void close()
+ {
+ future.cancel(false);
+ }
+ }
+
/**
* Processor manager that effectively delays future resolution by deferring
it through {@link #exec}.
* Especially useful on single-threaded test cases. This helps us ensure
that things work when channels don't have
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
