This is an automated email from the ASF dual-hosted git repository.
gian pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new b1878d6e731 Speed up MSQWorkerTaskLauncherRetryTest. (#18924)
b1878d6e731 is described below
commit b1878d6e731bf8a45271d79b63ea528e4267af73
Author: Gian Merlino <[email protected]>
AuthorDate: Fri Jan 16 22:52:43 2026 -0800
Speed up MSQWorkerTaskLauncherRetryTest. (#18924)
This patch takes it from 3 minutes to near instant. The issue was that
when a task is canceled during WorkerManager shutdown, the mock
OverlordClient ignored that, leading to a situation where the
WorkerManager had to time out (which takes a minute). In this patch,
the mock client reports the task as failed once it has been canceled.
---
.../apache/druid/msq/indexing/MSQWorkerTaskLauncherRetryTest.java | 8 ++++++--
1 file changed, 6 insertions(+), 2 deletions(-)
diff --git
a/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/MSQWorkerTaskLauncherRetryTest.java
b/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/MSQWorkerTaskLauncherRetryTest.java
index cecadc55fb9..5223dae9834 100644
---
a/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/MSQWorkerTaskLauncherRetryTest.java
+++
b/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/MSQWorkerTaskLauncherRetryTest.java
@@ -230,6 +230,7 @@ public class MSQWorkerTaskLauncherRetryTest
{
private final ConcurrentSkipListSet<Integer> unknownLocationWorkers = new
ConcurrentSkipListSet<>();
private final ConcurrentSkipListSet<Integer> failedWorkers = new
ConcurrentSkipListSet<>();
+ private final ConcurrentSkipListSet<String> canceledTasks = new
ConcurrentSkipListSet<>();
@Override
public ListenableFuture<URI> findCurrentLeader()
@@ -246,6 +247,7 @@ public class MSQWorkerTaskLauncherRetryTest
@Override
public ListenableFuture<Void> cancelTask(String taskId)
{
+ canceledTasks.add(taskId);
return Futures.immediateFuture(null);
}
@@ -265,7 +267,9 @@ public class MSQWorkerTaskLauncherRetryTest
final Map<String, TaskStatus> taskStatusMap = new HashMap<>();
for (String taskId : taskIds) {
int workerNumber = MSQTasks.workerFromTaskId(taskId);
- if (failedWorkers.contains(workerNumber)) {
+ if (canceledTasks.contains(taskId)) {
+ taskStatusMap.put(taskId, TaskStatus.failure(taskId, "Canceled"));
+ } else if (failedWorkers.contains(workerNumber)) {
taskStatusMap.put(taskId, TaskStatus.failure(taskId, "Task failed"));
} else if (unknownLocationWorkers.contains(workerNumber)) {
taskStatusMap.put(taskId,
TaskStatus.running(taskId).withLocation(TaskLocation.unknown()));
@@ -279,7 +283,7 @@ public class MSQWorkerTaskLauncherRetryTest
@Override
public ListenableFuture<TaskStatusResponse> taskStatus(String taskId)
{
- if (failedWorkers.contains(MSQTasks.workerFromTaskId(taskId))) {
+ if (canceledTasks.contains(taskId) ||
failedWorkers.contains(MSQTasks.workerFromTaskId(taskId))) {
return Futures.immediateFuture(new TaskStatusResponse(taskId,
createFailedTaskStatus(taskId)));
}
if (unknownLocationWorkers.contains(MSQTasks.workerFromTaskId(taskId))) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
