FrankChen021 commented on code in PR #20825:
URL: https://github.com/apache/druid/pull/20825#discussion_r4480525991
##########
multi-stage-query/src/main/java/org/apache/druid/msq/dart/worker/http/DartWorkerResource.java:
##########
@@ -120,19 +123,32 @@ public Response httpPostWorkOrder(
)
{
MSQResourceUtils.authorizeAdminRequest(permissionMapper, authorizerMapper,
req);
- final String controllerHost = req.getHeader(HEADER_CONTROLLER_HOST);
- if (controllerHost == null) {
- throw DruidException.forPersona(DruidException.Persona.DEVELOPER)
- .ofCategory(DruidException.Category.INVALID_INPUT)
- .build("Missing controllerId[%s]",
HEADER_CONTROLLER_HOST);
- }
-
- workerRunner.startWorker(queryId, controllerHost,
workOrder.getWorkerContext())
+ workerRunner.startWorker(queryId, getControllerHost(req),
workOrder.getWorkerContext())
.postWorkOrder(workOrder);
return Response.status(Response.Status.ACCEPTED).build();
}
+ /**
+ * Process a {@link SyncWorkersRequest} from {@link DartWorkerSyncer}.
+ */
+ @POST
+ @Consumes({MediaType.APPLICATION_JSON,
SmileMediaTypes.APPLICATION_JACKSON_SMILE})
+ @Produces(MediaType.APPLICATION_JSON)
+ @Path("/sync")
+ public SyncWorkersResponse httpPostSync(
+ final SyncWorkersRequest request,
+ @Context final HttpServletRequest req
+ )
+ {
+ MSQResourceUtils.authorizeAdminRequest(permissionMapper, authorizerMapper,
req);
+ if (request == null) {
Review Comment:
Rechecked at the current head and confirmed resolved: the null-body test now
expects INVALID_INPUT and verifies that the worker runner is untouched. No
further action is needed on this thread. Reviewed 3 of 36 changed files in this
reply-only follow-up.
<!-- mergelens:review -->
##########
multi-stage-query/src/main/java/org/apache/druid/msq/dart/controller/DartControllerRegistry.java:
##########
@@ -123,6 +132,24 @@ public void stop()
if (cleanupExec != null) {
cleanupExec.shutdown();
}
+
+ // Cancel any queries that are still running. This registry is stopped in
the NORMAL stage, after
+ // ANNOUNCEMENTS and SERVER, so by this point the Broker has been removed
from discovery and the
+ // druid.server.http.gracefulShutdownTimeout has already elapsed.
+ final List<ControllerHolder> holders = new
ArrayList<>(controllerMap.values());
+ if (!holders.isEmpty()) {
+ log.info("Canceling [%d] queries that are still running.",
holders.size());
+
+ // Cancel all controllers in sequence. It would be nicer to do this in
parallel, since it involves sending
+ // stopWorker calls to active workers. This would require some
restructuring of the code to return Futures,
+ // or an executor here used only for cancellation.
+ for (final ControllerHolder holder : holders) {
+ holder.cancel(CancellationReason.CONTROLLER_SHUTDOWN, null);
Review Comment:
Rechecked at the current head: this remains a valid P2. holder.cancel() is
still synchronous and waits for worker stop requests before
awaitControllerCompletion() begins, so deferring it leaves the stated 10-second
shutdown bound unenforced. Reviewed 3 of 36 changed files in this reply-only
follow-up.
<!-- mergelens:review -->
##########
multi-stage-query/src/main/java/org/apache/druid/msq/dart/worker/DartWorkerRunner.java:
##########
@@ -263,6 +360,54 @@ public void stop()
}
}
+ /**
+ * Handles a heartbeat from a controller. Called via {@link
PollListener#pollReceived(String)} by
+ * {@link MessageRelayResource#httpGetMessagesFromOutbox}.
+ */
+ public void controllerHeartbeat(final String controllerHost)
+ {
+ synchronized (this) {
+ lastControllerHeartbeat.put(controllerHost, clock.millis());
+ }
+ }
+
+ /**
+ * Cancels the {@link WorkerHolder} associated with controllers that have
not made contact in
+ * {@link DartWorkerConfig#getControllerTimeout()}. Called periodically in
{@link #controllerReaperExec}.
+ */
+ void reapControllersIfNeeded()
+ {
+ final Set<String> controllersToReap = new HashSet<>();
+
+ synchronized (this) {
+ final long now = clock.millis();
+ final Iterator<Map.Entry<String, Long>> it =
lastControllerHeartbeat.entrySet().iterator();
+ while (it.hasNext()) {
+ final Map.Entry<String, Long> entry = it.next();
+ if (now - entry.getValue() > controllerTimeoutMillis) {
+ controllersToReap.add(entry.getKey());
+ it.remove();
+ }
+ }
+
+ for (final String controllerHost : controllersToReap) {
+ log.warn(
+ "Reaping controller[%s] that has not made contact in over [%,d]
ms.",
+ controllerHost,
+ controllerTimeoutMillis
+ );
+ }
+
+ for (final WorkerHolder holder :
getWorkersForControllers(controllersToReap)) {
+ holder.runRef.cancel();
+ }
+
+ for (final String controllerHost : controllersToReap) {
+ outbox.resetOutbox(controllerHost);
Review Comment:
Rechecked at the current head and confirmed resolved. The controller relay
uses StandardRetryPolicy.unlimited(), so the reset-induced HTTP 500 is retried,
the retry observes the new epoch, and
EmbeddedDartLivenessTest#test_resetOutbox_relayRecovers covers recovery. No
further action is needed here. Reviewed 3 of 36 changed files in this
reply-only follow-up.
<!-- 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]
