abhishekrb19 commented on code in PR #16226:
URL: https://github.com/apache/druid/pull/16226#discussion_r1560198230
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/util/SqlStatementResourceHelper.java:
##########
@@ -361,22 +364,42 @@ public static MSQStagesReport.Stage
getFinalStage(MSQTaskReportPayload msqTaskRe
return null;
}
- public static Map<String, Object> getQueryExceptionDetails(Map<String,
Object> payload)
+ @Nullable
+ private static MSQErrorReport getQueryExceptionDetails(MSQTaskReportPayload
payload)
{
- return getMap(getMap(payload, "status"), "errorReport");
+ return payload == null ? null : payload.getStatus().getErrorReport();
}
- public static Map<String, Object> getMap(Map<String, Object> map, String key)
+ @Nullable
+ public static MSQTaskReportPayload getPayload(TaskReport.ReportMap reportMap)
{
- if (map == null) {
+ if (reportMap == null) {
return null;
}
- return (Map<String, Object>) map.get(key);
+
+ com.google.common.base.Optional<MSQTaskReport> report =
reportMap.findReport("multiStageQuery");
Review Comment:
Just an observation: it's slightly weird that we have `Optional` usages
from both `guava` and `java.util` packages in the same class.
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/sql/resources/SqlStatementResource.java:
##########
@@ -735,8 +735,8 @@ private Optional<Yielder<Object[]>> getResultYielder(
);
}
- MSQTaskReportPayload msqTaskReportPayload =
jsonMapper.convertValue(SqlStatementResourceHelper.getPayload(
- contactOverlord(overlordClient.taskReportAsMap(queryId), queryId)),
MSQTaskReportPayload.class);
+ MSQTaskReportPayload msqTaskReportPayload =
SqlStatementResourceHelper.getPayload(
+ contactOverlord(overlordClient.taskReportAsMap(queryId), queryId));
Review Comment:
nit: for readability
```suggestion
MSQTaskReportPayload msqTaskReportPayload =
SqlStatementResourceHelper.getPayload(
contactOverlord(overlordClient.taskReportAsMap(queryId), queryId)
);
```
##########
indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/SinglePhaseParallelIndexingTest.java:
##########
@@ -335,6 +338,38 @@ public void testRunInParallel()
testRunAndOverwrite(Intervals.of("2017-12/P1M"), Granularities.DAY);
}
+ @Test
+ public void testGetRunningTaskReports() throws Exception
+ {
+ final ParallelIndexSupervisorTask task = newTask(
+ Intervals.of("2017-12/P1M"),
+ Granularities.DAY,
+ false,
+ true
+ );
+ task.addToContext(Tasks.FORCE_TIME_CHUNK_LOCK_KEY, lockGranularity ==
LockGranularity.TIME_CHUNK);
+ task.addToContext(DISABLE_TASK_INJECT_CONTEXT_KEY, true);
+
+ // Keep tasks running until finish is triggered
+ getIndexingServiceClient().keepTasksRunning();
+ getIndexingServiceClient().runTask(task.getId(), task);
+
+ // Allow enough time for sub-tasks to be in running state
+ Thread.sleep(2000);
Review Comment:
I suppose adding a latch or a similar mechanism for testing would make it
more involved?
##########
server/src/test/java/org/apache/druid/client/indexing/NoopOverlordClient.java:
##########
@@ -84,7 +85,7 @@ public ListenableFuture<TaskPayloadResponse>
taskPayload(String taskId)
}
@Override
- public ListenableFuture<Map<String, Object>> taskReportAsMap(String taskId)
+ public ListenableFuture<TaskReport.ReportMap> taskReportAsMap(String taskId)
Review Comment:
I'd suggest adding a simple serde test that verifies that the old and new
objects being returned roundtrip fine for compatibility reasons (similar to
this comment https://github.com/apache/druid/pull/16217#discussion_r1545912070).
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/sql/resources/SqlStatementResource.java:
##########
@@ -746,8 +746,8 @@ private Optional<Yielder<Object[]>> getResultYielder(
} else if (msqControllerTask.getQuerySpec().getDestination() instanceof
DurableStorageMSQDestination) {
- MSQTaskReportPayload msqTaskReportPayload =
jsonMapper.convertValue(SqlStatementResourceHelper.getPayload(
- contactOverlord(overlordClient.taskReportAsMap(queryId), queryId)),
MSQTaskReportPayload.class);
+ MSQTaskReportPayload msqTaskReportPayload =
SqlStatementResourceHelper.getPayload(
+ contactOverlord(overlordClient.taskReportAsMap(queryId), queryId));
Review Comment:
```suggestion
MSQTaskReportPayload msqTaskReportPayload =
SqlStatementResourceHelper.getPayload(
contactOverlord(overlordClient.taskReportAsMap(queryId), queryId)
);
```
--
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]
