FrankChen021 commented on code in PR #20321:
URL: https://github.com/apache/druid/pull/20321#discussion_r3985530014
##########
multi-stage-query/src/main/java/org/apache/druid/msq/querykit/ChainedProcessorManager.java:
##########
@@ -88,32 +111,70 @@ public ChainedProcessorManager(
@Override
public ListenableFuture<Optional<ProcessorAndCallback<Object>>> next()
{
- if (closed) {
- throw new IllegalStateException();
- } else if (first != null) {
- Optional<ProcessorAndCallback<A>> processorAndCallbackOptional =
Futures.getUnchecked(first.next());
- if (processorAndCallbackOptional.isPresent()) {
- // More processors left to run.
- firstProcessorCount.incrementAndGet();
- ProcessorAndCallback<A> aProcessorAndCallback =
processorAndCallbackOptional.get();
- //noinspection unchecked
- return
Futures.immediateFuture(Optional.of((ProcessorAndCallback<Object>)
aProcessorAndCallback));
- } else {
- first = null;
- checkFirstProcessorComplete();
+ final ListenableFuture<Optional<ProcessorAndCallback<A>>> nextFromFirst;
+
+ synchronized (lock) {
+ if (closed) {
+ throw new IllegalStateException();
+ } else if (first == null) {
+ return nextFromRest();
}
+
+ nextFromFirst = first.next();
}
- //noinspection unchecked
+ return FutureUtils.transformAsync(
+ nextFromFirst,
+ processorAndCallback -> {
+ synchronized (lock) {
+ if (!closed) {
+ if (processorAndCallback.isPresent()) {
+ // More processors left to run.
+ firstProcessorCount++;
+ //noinspection unchecked
+ return
Futures.immediateFuture(Optional.of((ProcessorAndCallback<Object>)
processorAndCallback.get()));
+ } else {
+ first = null;
+ checkFirstProcessorComplete();
+ return nextFromRest();
+ }
+ }
+ }
+
+ // Closed while we were waiting. Clean up the processor we were
handed.
+ if (processorAndCallback.isPresent()) {
+ final FrameProcessor<A> processor =
processorAndCallback.get().processor();
+ CloseableUtils.closeAndSuppressExceptions(
+ processor::cleanup,
+ e -> log.noStackTrace().warn(e, "Failed to clean up
processor[%s] after close", processor)
+ );
+ }
+
+ return Futures.immediateFuture(Optional.empty());
+ }
+ );
+ }
+
+ /**
+ * Returns the next processor from {@link #restFuture}, once that manager is
available.
+ */
+ private ListenableFuture<Optional<ProcessorAndCallback<Object>>>
nextFromRest()
+ {
+ //noinspection unchecked, rawtypes
return FutureUtils.transformAsync(
Review Comment:
[P2] Clean up rest processors that arrive after close
Once restFuture has already been resolved, close() cannot cancel it; a
pending rest.next() can therefore still resolve with a ProcessorAndCallback
after the manager is closed. nextFromRest() forwards that value without
checking closed or cleaning up the returned FrameProcessor, unlike the
first-phase callback above. During a failure or cancellation, RunAllFullyWidget
can discard this late processor after its manager has been closed, leaving its
channels or other resources uncleaned. Route rest.next() through the same
post-close cleanup path (or otherwise cancel/track the in-flight future).
--
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]
