gianm commented on code in PR #20825:
URL: https://github.com/apache/druid/pull/20825#discussion_r4459115294
##########
multi-stage-query/src/main/java/org/apache/druid/msq/dart/worker/DartWorkerRunner.java:
##########
@@ -180,6 +234,57 @@ public void stopWorker(final String queryId)
}
}
+ /**
+ * Handles a sync from a controller.
+ *
+ * @see DartWorkerSyncer the controller side
+ */
+ public SyncWorkersResponse syncController(final String controllerHost, final
SyncWorkersRequest request)
+ {
+ final Set<String> queryIds = request.queryIds();
+ final List<WorkerHolder> holdersToCancel = new ArrayList<>();
+ final Set<String> runningQueryIds = new HashSet<>();
+
+ synchronized (this) {
+ final long now = clock.millis();
+ lastControllerHeartbeat.put(controllerHost, now);
+
+ for (final Map.Entry<String, WorkerHolder> entry : workerMap.entrySet())
{
+ final WorkerHolder holder = entry.getValue();
+ if (!holder.controllerHost.equals(controllerHost)) {
+ continue;
+ }
+
+ // Report the query as running even if we cancel it below, since it
has not exited yet.
+ runningQueryIds.add(entry.getKey());
+
+ if (queryIds.contains(entry.getKey())) {
+ // Null out missingFromSyncSince, signifying that this query is
meant to be active.
+ holder.missingFromSyncSince = null;
+ } else {
+ // Add this query to workersToCancel if it has been missing long
enough.
+ final Long missingFromSyncSince = holder.missingFromSyncSince;
+ if (missingFromSyncSince == null) {
+ holder.missingFromSyncSince = now;
+ } else if (now - missingFromSyncSince >= syncGraceMillis) {
+ holdersToCancel.add(holder);
+ }
+ }
+ }
+ }
+
+ for (final WorkerHolder holder : holdersToCancel) {
+ log.info(
+ "Canceling query[%s] because controller[%s] no longer lists it as
running.",
+ holder.workerContext.queryId(),
+ controllerHost
+ );
+ holder.runRef.cancel();
Review Comment:
What metric did you have in mind?
--
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]
