FrankChen021 commented on code in PR #20505:
URL: https://github.com/apache/druid/pull/20505#discussion_r4218899918
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -385,6 +417,37 @@ public String toString()
}
}
+ /**
+ * Represents a task that waits until all prior task groups that were
publishing
+ * to overlapping partitions have completed.
+ */
+ private class TaskWaitingToPublish
+ {
+ final CountDownLatch latch;
+ final SettableFuture<Boolean> readyToPublish = SettableFuture.create();
+ final Map<TaskGroup, Set<PartitionIdType>> blockingTaskGroups;
+
+ TaskWaitingToPublish(String taskId, Map<TaskGroup, Set<PartitionIdType>>
blockingTaskGroups)
+ {
+ this.latch = new CountDownLatch(blockingTaskGroups.size());
+ this.blockingTaskGroups = new HashMap<>(blockingTaskGroups);
+
+ tasksWaitingToPublish.put(taskId, this);
+
+ for (TaskGroup group : blockingTaskGroups.keySet()) {
+ group.addCompletionListener(() -> {
+ latch.countDown();
+ blockingTaskGroups.remove(group);
Review Comment:
[P2] Avoid mutating the map being iterated
**Finding:** The constructor iterates `blockingTaskGroups.keySet()` while
each completion callback removes from that same constructor argument. If a
prior group completes during registration, `addCompletionListener` can run the
callback inline (or concurrently), modifying the `HashMap` during iteration and
throwing `ConcurrentModificationException` before the readiness future is
returned.
**Suggestion:** Register listeners from an immutable snapshot and update a
separate thread-safe tracking structure from the callbacks.
##########
indexing-service/src/main/java/org/apache/druid/indexing/common/actions/SegmentTransactionalAppendAction.java:
##########
@@ -219,12 +222,31 @@ public SegmentPublishResult perform(Task task,
TaskActionToolbox toolbox)
}
IndexTaskUtils.emitSegmentPublishMetrics(retVal, task, toolbox);
+ return retVal;
+ }
- if (toolbox.shouldFailSegmentPublishImmediately(retVal, task,
supervisorId, startMetadata)) {
- return SegmentPublishResult.fail(retVal.getErrorMsg());
- } else {
- return retVal;
- }
+ @Override
+ public boolean canPerformAsync(Task task, TaskActionToolbox toolbox)
+ {
+ return supervisorId != null && startMetadata != null;
+ }
+
+ @Override
+ public ListenableFuture<SegmentPublishResult> performAsync(Task task,
TaskActionToolbox toolbox)
+ {
+ final ListenableFuture<Boolean> taskReadyToPublishFuture
+ = toolbox.isTaskReadyToPublish(task, supervisorId, startMetadata);
+ return Futures.transform(
+ taskReadyToPublishFuture,
+ taskIsReadyToPublish -> {
+ if (Boolean.TRUE.equals(taskIsReadyToPublish)) {
+ return
SegmentPublishResult.doNotRetryOffsetMismatchFailure(perform(task, toolbox));
+ } else {
+ return SegmentPublishResult.retryableFailure("Task is not ready to
publish yet");
+ }
+ },
+ MoreExecutors.directExecutor()
Review Comment:
[P2] Keep publish work off the supervisor notice thread
**Finding:** The readiness future is completed by `TaskGroup.onCompleted`
during `checkPendingCompletionTasks`, which runs on this supervisor's
single-threaded notice executor. With a direct executor, this transform calls
`perform` inline; that method enters the task-lock critical section and
performs metadata-store publication, so a slow publish blocks this supervisor
from processing subsequent status checks and notices. The same executor choice
is present in the insert action.
**Suggestion:** Dispatch the publish operation to a dedicated executor when
readiness completes, leaving the supervisor notice thread free to process its
queue.
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/http/OverlordResource.java:
##########
@@ -506,34 +515,50 @@ public Response getWorkerConfigHistory(
@Path("/action")
@Produces(MediaType.APPLICATION_JSON)
@ResourceFilters(StateResourceFilter.class)
- public Response doAction(final TaskActionHolder holder)
+ public void doAction(
+ final TaskActionHolder holder,
+ @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();
+ })
+ );
+ asyncContext.setTimeout(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(),
Map.of("result", result));
Review Comment:
[P1] Preserve null task-action results
**Finding:** Several existing actions, including `LockReleaseAction` and
`UpdateLocationAction`, implement `TaskAction<Void>` and return `null`.
`Map.of` rejects null values, so these successful actions throw from
`onSuccess` before `completeRequest` can send the result, leaving the async
request unanswered.
**Suggestion:** Build the response with a null-tolerant map so the existing
`result: null` response contract is preserved.
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/http/OverlordResource.java:
##########
@@ -884,4 +909,19 @@ private List<TaskStatusPlus> securedTaskStatusPlus(
)
);
}
+
+ private void completeRequest(AsyncContext context, int statusCode, Object
result)
+ {
+ try {
+ final HttpServletResponse response = (HttpServletResponse)
context.getResponse();
+ response.setStatus(statusCode);
+ if (result != null) {
+ response.setContentType(MediaType.APPLICATION_JSON);
+ jsonMapper.writeValue(response.getOutputStream(), result);
+ }
+ }
+ catch (Exception e) {
+ log.noStackTrace().warn(e, "Request timed out or is closed already");
+ }
+ }
Review Comment:
[P1] Complete the async response on every path
**Finding:** `doAction` starts an `AsyncContext`, but this helper only sets
the status and optionally writes a body; it never completes the context.
Consequently, even a finished task action leaves the HTTP exchange open until
the async timeout, and the no-leader path returns here before installing any
timeout listener.
**Suggestion:** Complete the async context after the response is written,
including when no response body is present.
--
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]
