FrankChen021 commented on code in PR #20505:
URL: https://github.com/apache/druid/pull/20505#discussion_r4228618538
##########
indexing-service/src/main/java/org/apache/druid/indexing/common/actions/LocalTaskActionClient.java:
##########
@@ -52,10 +58,28 @@ public <RetType> RetType submit(TaskAction<RetType>
taskAction)
log.debug("Performing action for task[%s]: %s", task.getId(), taskAction);
final long performStartTime = System.currentTimeMillis();
final RetType result = performAction(taskAction);
- emitTimerMetric("task/action/run/time", taskAction,
System.currentTimeMillis() - performStartTime);
+ emitTimerMetric(RUN_TIME_METRIC, taskAction, System.currentTimeMillis() -
performStartTime);
return result;
}
+ @Override
+ public <RetType> ListenableFuture<RetType> submitAsync(TaskAction<RetType>
taskAction)
+ {
+ if (taskAction.canPerformAsync(task, toolbox)) {
+ final Stopwatch actionRunTime = Stopwatch.createStarted();
+ return Futures.transform(
+ taskAction.performAsync(task, toolbox),
+ v -> {
+ emitTimerMetric(RUN_TIME_METRIC, taskAction,
actionRunTime.millisElapsed());
+ return v;
+ },
+ MoreExecutors.directExecutor()
+ );
+ } else {
+ return Futures.immediateFuture(submit(taskAction));
Review Comment:
[P1] Return synchronous action failures as failed futures
**Finding:** The fallback evaluates submit(taskAction) before
immediateFuture can be created, and the async branch can likewise throw while
evaluating canPerformAsync or performAsync. Those exceptions escape submitAsync
instead of reaching the future callback. OverlordResource has already called
startAsync() and invokes submitAsync before registering its timeout listener,
so an ordinary action failure can leave the HTTP request without a response,
timeout, or completion.
**Suggestion:** Convert synchronous failures from the async decision and
both execution paths into failed futures, or guard the call in the resource so
every failure completes the servlet context through the existing error callback.
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/http/OverlordResource.java:
##########
@@ -506,34 +515,57 @@ public Response getWorkerConfigHistory(
@Path("/action")
@Produces(MediaType.APPLICATION_JSON)
@ResourceFilters(StateResourceFilter.class)
- public Response doAction(final TaskActionHolder holder)
+ public void doAction(
+ final TaskActionHolder holder,
+ @Nullable @PathParam("timeout") final Long timeoutMillis,
+ @Context HttpServletRequest request
+ )
{
- return asLeaderWith(
- taskMaster.getTaskActionClient(holder.getTask()),
- new Function<>()
+ final AsyncContext asyncContext = request.startAsync();
+ final Optional<TaskActionClient> taskActionClient =
taskMaster.getTaskActionClient(holder.getTask());
+ if (!taskActionClient.isPresent()) {
+ // Encourage client to try again soon, when we'll likely have a redirect
set up
+ completeRequest(asyncContext,
Status.SERVICE_UNAVAILABLE.getStatusCode(), null);
+ return;
+ }
+
+ final ListenableFuture<?> future =
taskActionClient.get().submitAsync(holder.getAction());
+ asyncContext.addListener(
+ ServletResourceUtils.createAsyncTimeoutListener(event -> {
+ // HTTP 204 NO_CONTENT is sent to the client.
+ future.cancel(true);
+ event.getAsyncContext().complete();
+ })
+ );
+
+ // Use a default timeout of 5 minutes
+ asyncContext.setTimeout(Configs.valueOrDefault(timeoutMillis, 5 * 60_000));
+
+ Futures.addCallback(
+ future,
+ new FutureCallback<Object>()
{
@Override
- public Response apply(TaskActionClient taskActionClient)
+ public void onSuccess(Object result)
{
- final Map<String, Object> retMap;
-
- // It would be great to verify that this worker is actually
supposed to be running the task before
- // actually doing the action. Some ideas for how that could be
done would be using some sort of attempt_id
- // or token that gets passed around.
+ completeRequest(
+ asyncContext,
+ Status.OK.getStatusCode(),
+ result == null ? Map.of() : Map.of("result", result)
Review Comment:
[P2] Preserve the result field for null task-action results
**Finding:** The previous synchronous endpoint always serialized a result
property, including {"result":null} for Void actions. The async success path
now serializes {} when the action returns null. RemoteTaskActionClient
tolerates the missing property, but callers that rely on the stable response
shape will observe an API regression.
**Suggestion:** Use a null-tolerant response representation that retains the
existing result field for successful null-valued actions.
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/http/OverlordResource.java:
##########
@@ -506,34 +515,57 @@ public Response getWorkerConfigHistory(
@Path("/action")
@Produces(MediaType.APPLICATION_JSON)
@ResourceFilters(StateResourceFilter.class)
- public Response doAction(final TaskActionHolder holder)
+ public void doAction(
+ final TaskActionHolder holder,
+ @Nullable @PathParam("timeout") final Long timeoutMillis,
Review Comment:
[P1] Make the async action timeout cover publish waits
**Finding:** The /action route has no {timeout} path segment, so
@PathParam("timeout") is never populated and setTimeout always uses five
minutes. The supervisor's default completionTimeout is 30 minutes; a legitimate
pending task group that takes longer than five minutes causes the timeout
listener to cancel the action and complete the request before the readiness
future finishes. The worker then sees an I/O or empty-response failure, and the
publish retry path does not retry that timeout, so the new blocking behavior
can fail a task instead of waiting for the prior group.
**Suggestion:** Bind the timeout to a real request parameter or route and
propagate a deadline that covers the configured supervisor completion wait,
with an explicit retry-safe timeout response.
--
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]
