This is an automated email from the ASF dual-hosted git repository.
amaechler 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 63e3c2422df fix: Stop HttpRemoteTaskRunner from launching tasks killed
while pending (#20740)
63e3c2422df is described below
commit 63e3c2422df8a67435de1cd47d2651f2434a35b4
Author: Andreas Maechler <[email protected]>
AuthorDate: Wed Oct 28 09:56:57 2026 -0600
fix: Stop HttpRemoteTaskRunner from launching tasks killed while pending
(#20740)
* Stop HttpRemoteTaskRunner from launching tasks killed while pending
HttpRemoteTaskRunner.shutdown() only acted on RUNNING and COMPLETE work
items. A task killed while PENDING or PENDING_WORKER_ASSIGN stayed in
pendingTaskIds, so the pending task executor assigned it to the next free
worker even though TaskQueue had already marked it FAILED and released
its locks. It ran until the next storage sync dropped the queue entry and
the management loop shut it down again.
shutdown() now removes pending work items from pendingTaskIds under
statusLock and completes them with a FAILED status outside the lock. The
item stays in tasks as COMPLETE, so a worker that still starts an
in-flight assignment gets the task killed through the existing COMPLETE
case in taskAddedOrUpdated(), and the next shutdown() call removes it.
* Only complete strict PENDING items on HttpRemoteTaskRunner shutdown
Completing a PENDING_WORKER_ASSIGN item from shutdown() races with the
assigning thread, which may revert the item to PENDING outside
statusLock and leave it PENDING with a completed future. shutdown() now
only removes such an item from pendingTaskIds so it cannot be assigned
again. If the worker starts the task, the next shutdown() call kills it
on the worker; if the assignment fails, the item goes back to PENDING
and the next shutdown() call completes it.
Also checks the failure reason in the pending task test, makes its
locals final and stops the task runner.
---
.../overlord/hrtr/HttpRemoteTaskRunner.java | 14 +++
.../overlord/hrtr/HttpRemoteTaskRunnerTest.java | 137 ++++++++++++++++++++-
2 files changed, 150 insertions(+), 1 deletion(-)
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/hrtr/HttpRemoteTaskRunner.java
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/hrtr/HttpRemoteTaskRunner.java
index 24f0fe12752..2378c737763 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/hrtr/HttpRemoteTaskRunner.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/hrtr/HttpRemoteTaskRunner.java
@@ -1246,6 +1246,7 @@ public class HttpRemoteTaskRunner implements
WorkerTaskRunner, TaskLogStreamer,
}
WorkerHolder workerHolderRunningTask = null;
+ HttpRemoteTaskRunnerWorkItem pendingWorkItem = null;
synchronized (statusLock) {
log.info("Shutdown [%s] because: [%s]", taskId, reason);
HttpRemoteTaskRunnerWorkItem taskRunnerWorkItem = tasks.get(taskId);
@@ -1255,6 +1256,14 @@ public class HttpRemoteTaskRunner implements
WorkerTaskRunner, TaskLogStreamer,
if (workerHolderRunningTask == null) {
log.info("Can't shutdown! No worker running task[%s]", taskId);
}
+ } else if (taskRunnerWorkItem.getState() ==
HttpRemoteTaskRunnerWorkItem.State.PENDING) {
+ // Removing it here stops the pending task executor from picking it
up before it is completed below.
+ pendingTaskIds.remove(taskId);
+ pendingWorkItem = taskRunnerWorkItem;
+ } else if (taskRunnerWorkItem.getState() ==
HttpRemoteTaskRunnerWorkItem.State.PENDING_WORKER_ASSIGN) {
+ // The thread assigning this task owns the work item. Once the
worker runs the task, or the item goes back
+ // to PENDING, a later shutdown request handles it.
+ pendingTaskIds.remove(taskId);
} else if (taskRunnerWorkItem.getState() ==
HttpRemoteTaskRunnerWorkItem.State.COMPLETE) {
tasks.remove(taskId);
}
@@ -1263,6 +1272,11 @@ public class HttpRemoteTaskRunner implements
WorkerTaskRunner, TaskLogStreamer,
}
}
+ if (pendingWorkItem != null) {
+ // taskComplete(..) must be called outside of statusLock, see comments
on method.
+ taskComplete(pendingWorkItem, null, TaskStatus.failure(taskId, reason));
+ }
+
//shutdown is called outside of lock as we don't want to hold the lock
while sending http request
//to worker.
if (workerHolderRunningTask != null) {
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/hrtr/HttpRemoteTaskRunnerTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/hrtr/HttpRemoteTaskRunnerTest.java
index 6d965b529b9..58f053eb0b2 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/hrtr/HttpRemoteTaskRunnerTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/hrtr/HttpRemoteTaskRunnerTest.java
@@ -22,6 +22,7 @@ package org.apache.druid.indexing.overlord.hrtr;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.base.Optional;
import com.google.common.base.Predicates;
+import com.google.common.collect.ForwardingMap;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
@@ -90,6 +91,7 @@ import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.atomic.AtomicInteger;
@@ -1785,6 +1787,7 @@ public class HttpRemoteTaskRunnerTest
.collect(Collectors.toSet())
.contains(pendingTask.getId())
);
+ Assertions.assertTrue(taskRunner.getPendingTasks().isEmpty());
Task completedTask = NoopTask.create();
taskRunner.run(completedTask);
@@ -1793,7 +1796,13 @@ public class HttpRemoteTaskRunnerTest
TaskStatus.success(completedTask.getId()),
TaskLocation.create("worker", 1, 2)
), workerHolder);
- Assertions.assertEquals(completedTask.getId(),
Iterables.getOnlyElement(taskRunner.getCompletedTasks()).getTaskId());
+ Assertions.assertEquals(
+ ImmutableSet.of(pendingTask.getId(), completedTask.getId()),
+ taskRunner.getCompletedTasks()
+ .stream()
+ .map(TaskRunnerWorkItem::getTaskId)
+ .collect(Collectors.toSet())
+ );
TestDruidNodeDiscovery druidNodeDiscovery = new TestDruidNodeDiscovery();
DruidNodeDiscoveryProvider druidNodeDiscoveryProvider =
EasyMock.createMock(DruidNodeDiscoveryProvider.class);
EasyMock.expect(druidNodeDiscoveryProvider.getForService(WorkerNodeService.DISCOVERY_SERVICE_KEY))
@@ -1812,6 +1821,132 @@ public class HttpRemoteTaskRunnerTest
}
+ @Timeout(60)
+ @Test
+ public void testShutdownPendingTaskIsNotAssignedToWorker() throws Exception
+ {
+ final TestDruidNodeDiscovery druidNodeDiscovery = new
TestDruidNodeDiscovery();
+ final DruidNodeDiscoveryProvider druidNodeDiscoveryProvider =
EasyMock.createMock(DruidNodeDiscoveryProvider.class);
+
EasyMock.expect(druidNodeDiscoveryProvider.getForService(WorkerNodeService.DISCOVERY_SERVICE_KEY))
+ .andReturn(druidNodeDiscovery)
+ .times(2);
+ EasyMock.replay(druidNodeDiscoveryProvider);
+
+ final HttpRemoteTaskRunner taskRunner = newHttpTaskRunnerInstance(
+ druidNodeDiscoveryProvider,
+ new NoopProvisioningStrategy<>()
+ );
+ taskRunner.start();
+ try {
+ // No workers yet, so the task stays pending
+ final Task killedTask = NoopTask.create();
+ final Future<TaskStatus> killedTaskFuture = taskRunner.run(killedTask);
+ taskRunner.shutdown(killedTask.getId(), "Killed while pending");
+
+ final Task otherTask = NoopTask.create();
+ final Future<TaskStatus> otherTaskFuture = taskRunner.run(otherTask);
+
+ druidNodeDiscovery.getListeners().get(0).nodesAdded(ImmutableList.of(
+ new DiscoveryDruidNode(
+ new DruidNode("service", "host1", false, 8080, null, true,
false),
+ NodeRole.MIDDLE_MANAGER,
+ ImmutableMap.of(
+ WorkerNodeService.DISCOVERY_SERVICE_KEY,
+ new WorkerNodeService("ip1", 1, "0",
WorkerConfig.DEFAULT_CATEGORY)
+ )
+ )
+ ));
+
+ Assertions.assertTrue(otherTaskFuture.get().isSuccess());
+ final TaskStatus killedTaskStatus = killedTaskFuture.get();
+ Assertions.assertEquals(TaskState.FAILED,
killedTaskStatus.getStatusCode());
+ Assertions.assertEquals("Killed while pending",
killedTaskStatus.getErrorMsg());
+ }
+ finally {
+ taskRunner.stop();
+ }
+ }
+
+ @Timeout(60)
+ @Test
+ public void testShutdownTaskBeingAssignedLeavesItToAssigningThread() throws
Exception
+ {
+ final TestDruidNodeDiscovery druidNodeDiscovery = new
TestDruidNodeDiscovery();
+ final DruidNodeDiscoveryProvider druidNodeDiscoveryProvider =
EasyMock.createMock(DruidNodeDiscoveryProvider.class);
+
EasyMock.expect(druidNodeDiscoveryProvider.getForService(WorkerNodeService.DISCOVERY_SERVICE_KEY))
+ .andReturn(druidNodeDiscovery)
+ .times(2);
+ EasyMock.replay(druidNodeDiscoveryProvider);
+
+ final Task task = NoopTask.create();
+ final CountDownLatch assignmentStarted = new CountDownLatch(1);
+ final CountDownLatch assignmentAllowed = new CountDownLatch(1);
+ // The test worker looks the task up here while assigning it, so this
holds the task in PENDING_WORKER_ASSIGN
+ final Map<Task, List<TaskAnnouncement>> toBeAssignedTasks = new
ForwardingMap<Task, List<TaskAnnouncement>>()
+ {
+ @Override
+ protected Map<Task, List<TaskAnnouncement>> delegate()
+ {
+ return ImmutableMap.of(
+ task,
+ ImmutableList.of(TaskAnnouncement.create(task,
TaskStatus.running(task.getId()), TaskLocation.unknown()))
+ );
+ }
+
+ @Override
+ public boolean containsKey(Object key)
+ {
+ assignmentStarted.countDown();
+ try {
+ assignmentAllowed.await();
+ }
+ catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ return super.containsKey(key);
+ }
+ };
+
+ final HttpRemoteTaskRunner taskRunner = newHttpTaskRunnerInstance(
+ druidNodeDiscoveryProvider,
+ new NoopProvisioningStrategy<>(),
+ EasyMock.createNiceMock(HttpClient.class),
+ toBeAssignedTasks
+ );
+ taskRunner.start();
+ try {
+ druidNodeDiscovery.getListeners().get(0).nodesAdded(ImmutableList.of(
+ new DiscoveryDruidNode(
+ new DruidNode("service", "host1", false, 8080, null, true,
false),
+ NodeRole.MIDDLE_MANAGER,
+ ImmutableMap.of(
+ WorkerNodeService.DISCOVERY_SERVICE_KEY,
+ new WorkerNodeService("ip1", 1, "0",
WorkerConfig.DEFAULT_CATEGORY)
+ )
+ )
+ ));
+
+ final Future<TaskStatus> future = taskRunner.run(task);
+ assignmentStarted.await();
+ taskRunner.shutdown(task.getId(), "Killed while being assigned");
+
+ Assertions.assertFalse(future.isDone());
+
Assertions.assertFalse(taskRunner.getPendingTasksList().contains(task.getId()));
+
+ // The assignment goes through, so the next shutdown request finds the
task RUNNING on the worker
+ assignmentAllowed.countDown();
+ while (taskRunner.getRunningTasks().isEmpty()) {
+ Thread.sleep(10);
+ }
+ Assertions.assertEquals(task.getId(),
Iterables.getOnlyElement(taskRunner.getRunningTasks()).getTaskId());
+ Assertions.assertFalse(future.isDone());
+ }
+ finally {
+ assignmentAllowed.countDown();
+ taskRunner.stop();
+ }
+ }
+
@Timeout(60)
@Test
public void testSyncMonitoring_finiteIteration()
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
