FrankChen021 commented on code in PR #20505:
URL: https://github.com/apache/druid/pull/20505#discussion_r4258993566
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/http/OverlordResource.java:
##########
@@ -506,34 +518,55 @@ public Response getWorkerConfigHistory(
@Path("/action")
@Produces(MediaType.APPLICATION_JSON)
@ResourceFilters(StateResourceFilter.class)
- public Response doAction(final TaskActionHolder holder)
+ public void performTaskAction(
+ final TaskActionHolder holder,
+ @Context HttpServletRequest request
+ )
{
- return asLeaderWith(
- taskMaster.getTaskActionClient(holder.getTask()),
- new Function<>()
+ final AsyncRequest asyncRequest = new AsyncRequest(request);
+ 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
+ asyncRequest.complete(Status.SERVICE_UNAVAILABLE.getStatusCode(), null);
+ return;
+ }
+
+ final ListenableFuture<?> future =
taskActionClient.get().submitAsync(holder.getAction());
+ asyncRequest.context.addListener(
+ ServletResourceUtils.createAsyncTimeoutListener(_ -> {
+ asyncRequest.complete(HttpResponseStatus.GATEWAY_TIMEOUT.code(),
null);
+ future.cancel(true);
+ })
+ );
+
+ // Use a default timeout of 15 minutes
+ asyncRequest.context.setTimeout(15 * 60_000);
+
+ Futures.addCallback(
+ future,
+ new FutureCallback<Object>()
{
@Override
- public Response apply(TaskActionClient taskActionClient)
+ public void onSuccess(Object result)
{
- final Map<String, Object> retMap;
+ // Use null-safe map since some task actions may return null result
+ final Map<String, Object> payload = new HashMap<>();
+ payload.put("result", result);
- // 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.
+ asyncRequest.complete(Status.OK.getStatusCode(), payload);
+ }
- try {
- final Object ret = taskActionClient.submit(holder.getAction());
- retMap = new HashMap<>();
- retMap.put("result", ret);
- }
- catch (Exception e) {
- log.warn(e, "Failed to perform task action");
- return Response.serverError().entity(ImmutableMap.of("error",
e.getMessage())).build();
+ @Override
+ public void onFailure(Throwable t)
+ {
+ if (t instanceof DruidException druidException) {
+ asyncRequest.complete(druidException.getStatusCode(),
druidException.toErrorResponse());
+ } else {
+
asyncRequest.complete(Status.INTERNAL_SERVER_ERROR.getStatusCode(), null);
Review Comment:
[P2] Generic async failures lose the error payload
**Finding:** Defect: the generic failure branch completes a bare 500
response, whereas the previous endpoint returned an error entity. Trigger: any
action failure that is not delivered as a direct DruidException, including
RuntimeException wrappers from transactional publish actions. Impact: remote
task clients and operators receive an empty 500 and lose the actionable message
and diagnostics previously exposed.
**Suggestion:** Preserve the previous JSON error payload and exception
logging in the generic failure branch while retaining the
DruidException-specific status handling.
##########
indexing-service/src/main/java/org/apache/druid/indexing/common/actions/TaskActionToolbox.java:
##########
@@ -140,25 +154,46 @@ public boolean canBatchSegmentAllocation()
}
/**
- * Checks if the given publish action should be failed without allowing any
- * more retries. A failed publish action should be retried only if there is
- * another task waiting to publish offsets for an overlapping set of
partitions.
+ * Performs the segment publish action when the given task is unblocked for
publish.
+ * A streaming task must wait for previously created tasks that are
+ * yet to publish offsets for an overlapping set of partitions.
*/
- public boolean shouldFailSegmentPublishImmediately(
- SegmentPublishResult result,
+ public ListenableFuture<SegmentPublishResult> publishSegmentsWhenReady(
Task task,
String supervisorId,
- DataSourceMetadata startMetadata
+ DataSourceMetadata startMetadata,
+ BiFunction<Task, TaskActionToolbox, SegmentPublishResult> publishAction
)
{
- if (result.isSuccess() || !result.isRetryable() || startMetadata == null) {
- return false;
+ // Try publishing later if the failure was due to offset mismatch
+ final ListenableFuture<Boolean> taskReadyToPublishFuture;
+ if (startMetadata == null || supervisorId == null) {
+ taskReadyToPublishFuture = Futures.immediateFuture(true);
+ } else {
+ taskReadyToPublishFuture =
supervisorManager.isTaskReadyToPublishSegments(
+ supervisorId,
+ task.getId(),
+ startMetadata
+ );
}
- return !getSupervisorManager().isAnotherTaskGroupPublishingToPartitions(
- supervisorId,
- task.getId(),
- startMetadata
+ return Futures.transform(
Review Comment:
[P1] Publish waits can fail local tasks after five minutes
**Finding:** Defect: this future can remain pending until prior task groups
finish, but local callers invoke it through LocalTaskActionClient.submit, whose
async branch waits only five minutes. Trigger: a predecessor remains pending or
the publish itself exceeds five minutes; the default supervisor completion
timeout is 30 minutes, and SequenceMetadata uses synchronous submit for this
action. Impact: the local-runner streaming task receives a wrapped
TimeoutException and fails instead of waiting for the predecessor or returning
the intended retryable publish result.
**Suggestion:** Give streaming publish waits a deadline aligned with the
supervisor completion window, or keep them out of the generic five-minute
async-action path so the task can receive the publish result or retry status.
--
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]
