FrankChen021 commented on code in PR #20290:
URL: https://github.com/apache/druid/pull/20290#discussion_r3968447402
##########
processing/src/main/java/org/apache/druid/query/ForwardingQueryProcessingPool.java:
##########
@@ -63,22 +64,54 @@ public <T, V> ListenableFuture<T>
submitRunnerTask(PrioritizedQueryRunnerCallabl
return delegate().submit(task);
}
+ /**
+ * The timeout only starts counting once a processing thread has actually
picked the task off the pool's queue,
+ * not when it is submitted. Otherwise a task that sits in the queue behind
other segments could exhaust its
+ * per-segment timeout without ever having been given a chance to run.
+ */
@Override
public <T, V> ListenableFuture<T> submitRunnerTask(
PrioritizedQueryRunnerCallable<T, V> task,
long timeout,
TimeUnit unit
)
{
- if (timeoutService != null) {
- return Futures.withTimeout(
- delegate().submit(task),
- timeout,
- unit,
- timeoutService
- );
+ if (timeoutService == null) {
+ return submitRunnerTask(task);
}
- return submitRunnerTask(task);
+
+ final SettableFuture<Void> started = SettableFuture.create();
+ final ListenableFuture<T> execFuture = submitRunnerTask(
+ new AbstractPrioritizedQueryRunnerCallable<T, V>(task.getPriority(),
task.getRunner())
+ {
+ @Override
+ public T call() throws Exception
+ {
+ started.set(null);
+ return task.call();
+ }
+ }
+ );
+ // If the task never gets to run (cancelled or rejected while queued),
unblock the transform below so that the
+ // returned future completes with the underlying outcome instead of
hanging forever.
+ execFuture.addListener(() -> started.set(null),
MoreExecutors.directExecutor());
Review Comment:
[P2] Queued futures can hang after shutdownNow
This listener is the only path that releases `started` when the wrapper
never enters `call()`. However, `PrioritizedExecutorService.shutdownNow()`
delegates to `ThreadPoolExecutor.shutdownNow()`, which drains queued
`FutureTask`s without invoking `cancel` or `run`; a queued `execFuture` can
therefore remain incomplete and this listener never runs. The returned
`timedFuture` then has no timeout scheduled and stays pending indefinitely, so
a caller waiting on the query's `allAsList` can hang during a forced
processing-pool shutdown. Ensure tasks discarded by executor shutdown
complete/cancel the exposed future or otherwise signal this gate.
--
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]
