FrankChen021 commented on code in PR #20313:
URL: https://github.com/apache/druid/pull/20313#discussion_r3989285338
##########
multi-stage-query/src/main/java/org/apache/druid/msq/dart/worker/DartWorkerClientImpl.java:
##########
@@ -140,8 +147,26 @@ public void close()
}
@Override
- public ListenableFuture<?> stopWorker(String workerId)
+ public ListenableFuture<Void> postWorkOrder(final String workerId, final
WorkOrder workOrder)
+ {
+ final ListenableFuture<Void> future = super.postWorkOrder(workerId,
workOrder);
Review Comment:
Confirmed fixed in `9b7eafa`: the stop-state check, `/workOrder` dispatch,
and active-work-order registration are now serialized under `clientMap`, so the
original late-`/workOrder` race is resolved. I rechecked all 5 changed files
and found no remaining issue with that specific finding.
<!-- mergelens:review -->
##########
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
`RequestTrackingClient.asyncRequest` invokes `delegate.asyncRequest` before
adding its future to `activeRequests`. For `postFinish`, fetches, or
`stopWorker`, another thread can enter `close()` after the delegate has started
the request but before this add; `close()` then snapshots and clears the set
and closes the locators, while the future is added afterward and is never
canceled. An in-flight request can therefore survive query teardown (and this
is another way for a stop request to be lost). Register the future under the
same lifecycle lock or use a tracker that checks a closed flag after
registration, and test this race.
##########
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
`close()` now cancels every tracked future, including the `/stop` future
created by `stopWorker`. This can race with controller cancellation:
`ControllerHolder.cancel()` calls `ControllerImpl.stop()` on one thread, which
sets `DartWorkerManager` to STOPPED and waits for its stop future, while the
controller loop processes the queued cancellation, sees STOPPED, and exits
through `Closer`, invoking this `close()` before the first `stop(true)`
returns. The canceled future makes `successfulAsList` finish without an
acknowledgement (and a still-scheduled request may never be sent), defeating
the manager's unlimited retry; its comment explicitly says a lost stop leaves a
worker zombie until server restart. Coordinate close with the stop wait or
exempt/drain `/stop` requests, and add a regression test for this concurrent
path.
--
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]
