gianm commented on code in PR #20825:
URL: https://github.com/apache/druid/pull/20825#discussion_r4472986094
##########
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:
I'd prefer to leave this for later, because as mentioned previously, it
would add complexity to the code and I'm not sure it's worth it.
##########
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:
I don't think this happens. A canceled long poll comes back to the Broker as
an HTTP error, which would be retried, and the retry would see the new outbox
epoch. The new test `EmbeddedDartLivenessTest#test_resetOutbox_relayRecovers`
should cover this.
--
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]
