FrankChen021 commented on code in PR #20293:
URL: https://github.com/apache/druid/pull/20293#discussion_r4024175561
##########
multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerHolder.java:
##########
@@ -185,11 +209,22 @@ public ListenableFuture<?> runAsync(
}
}
- updateStateOnQueryComplete(reportListener.getReport());
+ reportMap = controller.finalReport();
+ if (reportMap != null) {
+ final TaskReport taskReport =
reportMap.get(MSQTaskReport.REPORT_KEY);
+ if (taskReport instanceof MSQTaskReport) {
+ final MSQTaskReportPayload report = ((MSQTaskReport)
taskReport).getPayload();
+ if (report != null) {
+ updateStateOnQueryComplete(report);
+ }
+ }
+ }
} else {
// Canceled before running.
+ final MSQTaskReportPayload canceledReport =
makeCanceledReport(cancelReason);
+ reportMap = TaskReport.buildTaskReports(new
MSQTaskReport(controller.queryId(), canceledReport));
synchronized (this) {
- reportListener.onQueryComplete(makeCanceledReport(cancelReason));
+ listener.onQueryComplete(canceledReport);
Review Comment:
Fixed in `1fa600454a`. The pre-run cancellation branch now snapshots the
canceled report under the holder lock and invokes `onQueryComplete` after
releasing it, allowing `getReports()` to acquire the lock while the callback is
still running. The regression test now performs the registry lookup on a
separate controller-pool thread and requires it to complete before releasing
the callback. Focused validation: 24 tests passed with 0 failures or errors.
##########
multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerHolder.java:
##########
@@ -169,7 +181,15 @@ public ListenableFuture<?> runAsync(
Thread.currentThread().setName(makeThreadName());
try {
- final CaptureReportQueryListener reportListener = new
CaptureReportQueryListener(listener);
+ final CaptureReportQueryListener reportListener = new
CaptureReportQueryListener(listener)
+ {
+ @Override
+ public void onQueryComplete(final MSQTaskReportPayload report)
+ {
+ finalReport = TaskReport.buildTaskReports(new
MSQTaskReport(controller.queryId(), report));
Review Comment:
Thanks, understood that returning the final report from `/liveReports` would
be acceptable. I kept `liveReports()` and `finalReport()` separate in this PR
so the live-snapshot API and the final-only publication signal remain explicit,
and so this concurrency fix does not also change the endpoint contract. The
duplicate `CaptureReportQueryListener` capture has still been removed; the
final report is retained only by `ControllerImpl`.
--
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]
