gianm commented on code in PR #20321:
URL: https://github.com/apache/druid/pull/20321#discussion_r4472394098
##########
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:
Fixed this by adjusting the contract so `RunAllFullyWidget` handles the
closing.
--
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]
