FrankChen021 commented on code in PR #20313:
URL: https://github.com/apache/druid/pull/20313#discussion_r4005004739
##########
multi-stage-query/src/main/java/org/apache/druid/msq/dart/worker/DartWorkerClientImpl.java:
##########
@@ -127,6 +138,13 @@ public void closeClient(final String workerIdString)
public void close()
{
synchronized (clientMap) {
+ closed = true;
+
+ // Cancel requests before closing locators so in-flight requests do not
continue retrying.
+ final List<ListenableFuture<?>> requests = new
ArrayList<>(activeRequests);
+ activeRequests.clear();
+ requests.forEach(request -> request.cancel(true));
Review Comment:
[P1] Do not cancel stop commands during close
**Finding:** close() cancels every tracked future, including the /stop
future created by stopWorker(). Controller cancellation can run stop(true) on
one thread while the controller loop exits through Closer on another thread;
canceling this future makes DartWorkerManager's successfulAsList finish without
an acknowledgement and can prevent the retrying stop request from being sent.
That defeats the manager's zombie-worker protection and can leave a canceled
query running on a worker.
**Suggestion:** Coordinate close with the stop wait or keep /stop requests
alive until DartWorkerManager has finished waiting, and add a regression test
for this concurrent lifecycle path.
##########
multi-stage-query/src/main/java/org/apache/druid/msq/dart/worker/DartWorkerClientImpl.java:
##########
@@ -184,6 +223,39 @@ private Pair<ServiceClient, Closeable>
getClientAndLocator(final String workerId
}
}
+ private static class RequestTrackingClient implements ServiceClient
+ {
+ private final ServiceClient delegate;
+ private final Set<ListenableFuture<?>> activeRequests;
+
+ private RequestTrackingClient(
+ final ServiceClient delegate,
+ final Set<ListenableFuture<?>> activeRequests
+ )
+ {
+ this.delegate = delegate;
+ this.activeRequests = activeRequests;
+ }
+
+ @Override
+ public <IntermediateType, FinalType> ListenableFuture<FinalType>
asyncRequest(
+ final RequestBuilder requestBuilder,
+ final HttpResponseHandler<IntermediateType, FinalType> handler
+ )
+ {
+ final ListenableFuture<FinalType> future =
delegate.asyncRequest(requestBuilder, handler);
+ activeRequests.add(future);
Review Comment:
[P2] Register requests atomically with close
**Finding:** RequestTrackingClient.asyncRequest() invokes the delegate
before adding its future to activeRequests. If close() acquires clientMap
between those operations, it snapshots and clears the set and closes the
locators, then this future is added afterward and is never canceled. An
in-flight postFinish, fetch, or stop request can therefore survive query
teardown, and a stop request can be lost or continue after the client is closed.
**Suggestion:** Register the future under the same lifecycle lock or make
registration recheck the closed state and cancel immediately, then cover the
interleaving with a concurrency test.
--
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]
