FrankChen021 commented on code in PR #20321:
URL: https://github.com/apache/druid/pull/20321#discussion_r4480650607
##########
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:
Follow-up review: this addresses my cleanup concern.
RunAllFullyWidget.cleanupNextProcessor() now tracks the outstanding next()
future before processorManager.close(), so a ProcessorAndCallback that resolves
after close is cleaned instead of discarded. The added regression test covers
this failure path. Reviewed 5 of 5 changed files.
<!-- mergelens:review -->
--
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]
