This is an automated email from the ASF dual-hosted git repository.
kfaraz 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 8da9a1b1071 minor: Pass full Task payload in TaskRunnerListener and
not just taskId (#20417)
8da9a1b1071 is described below
commit 8da9a1b1071327beff5d7baba9c49f68b84420e2
Author: Kashif Faraz <[email protected]>
AuthorDate: Mon Sep 28 17:15:06 2026 +0530
minor: Pass full Task payload in TaskRunnerListener and not just taskId
(#20417)
Changes:
- Pass `Task` payload in `TaskRunnerListener` methods `locationChanged` and
`statusChanged`
- This allows listeners (particularly in extensions) to take better
decisions on these notifications
as they can now extract the task type, datasource and other info of the
tasks.
- Fix `HttpRemoteTaskRunner` to fetch both `TaskStatus` and `Task` payload
from metadata store
when it discovers a new task from one of the workers
---
.../druid/k8s/overlord/KubernetesTaskRunner.java | 4 +-
.../overlord/MultipleKubernetesTaskRunnerTest.java | 4 +-
.../compact/OverlordCompactionScheduler.java | 13 ++--
.../overlord/BaseRestorableTaskRunner.java | 10 ++-
.../druid/indexing/overlord/ForkingTaskRunner.java | 22 ++++++-
.../overlord/SingleTaskBackgroundRunner.java | 14 ++---
.../indexing/overlord/TaskRunnerListener.java | 5 +-
.../druid/indexing/overlord/TaskRunnerUtils.java | 23 ++++---
.../indexing/overlord/ThreadingTaskRunner.java | 22 ++++++-
.../overlord/hrtr/HttpRemoteTaskRunner.java | 29 +++++----
.../scheduledbatch/ScheduledBatchTaskManager.java | 7 ++-
.../supervisor/SeekableStreamSupervisor.java | 4 +-
.../druid/indexing/worker/WorkerTaskManager.java | 6 +-
.../overlord/SingleTaskBackgroundRunnerTest.java | 9 +--
.../druid/indexing/overlord/TestTaskRunner.java | 14 ++---
.../overlord/hrtr/HttpRemoteTaskRunnerTest.java | 71 ++++++++++++++++++++--
16 files changed, 185 insertions(+), 72 deletions(-)
diff --git
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesTaskRunner.java
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesTaskRunner.java
index fa7148a6151..9d7254d193d 100644
---
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesTaskRunner.java
+++
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesTaskRunner.java
@@ -333,13 +333,13 @@ public class KubernetesTaskRunner implements
TaskLogStreamer, TaskRunner
@Override
public void updateStatus(Task task, TaskStatus status)
{
- TaskRunnerUtils.notifyStatusChanged(listeners, task.getId(), status);
+ TaskRunnerUtils.notifyStatusChanged(listeners, task, status);
}
@Override
public void updateLocation(Task task, TaskLocation location)
{
- TaskRunnerUtils.notifyLocationChanged(listeners, task.getId(), location);
+ TaskRunnerUtils.notifyLocationChanged(listeners, task, location);
}
@Override
diff --git
a/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/MultipleKubernetesTaskRunnerTest.java
b/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/MultipleKubernetesTaskRunnerTest.java
index 2033b23fb60..26212181cd7 100644
---
a/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/MultipleKubernetesTaskRunnerTest.java
+++
b/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/MultipleKubernetesTaskRunnerTest.java
@@ -1069,13 +1069,13 @@ public class MultipleKubernetesTaskRunnerTest extends
EasyMockSupport
}
@Override
- public void locationChanged(String taskId, TaskLocation newLocation)
+ public void locationChanged(Task task, TaskLocation newLocation)
{
}
@Override
- public void statusChanged(String taskId, TaskStatus status)
+ public void statusChanged(Task task, TaskStatus status)
{
}
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/compact/OverlordCompactionScheduler.java
b/indexing-service/src/main/java/org/apache/druid/indexing/compact/OverlordCompactionScheduler.java
index c2a0dabb65a..647b5774513 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/compact/OverlordCompactionScheduler.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/compact/OverlordCompactionScheduler.java
@@ -29,6 +29,7 @@ import org.apache.druid.error.DruidException;
import org.apache.druid.indexer.TaskLocation;
import org.apache.druid.indexer.TaskStatus;
import org.apache.druid.indexing.common.actions.TaskActionClientFactory;
+import org.apache.druid.indexing.common.task.Task;
import org.apache.druid.indexing.overlord.GlobalTaskLockbox;
import org.apache.druid.indexing.overlord.TaskMaster;
import org.apache.druid.indexing.overlord.TaskQueryTool;
@@ -198,16 +199,16 @@ public class OverlordCompactionScheduler implements
CompactionScheduler
}
@Override
- public void locationChanged(String taskId, TaskLocation newLocation)
+ public void locationChanged(Task task, TaskLocation newLocation)
{
// Do nothing
}
@Override
- public void statusChanged(String taskId, TaskStatus status)
+ public void statusChanged(Task task, TaskStatus status)
{
if (status.isComplete()) {
- onTaskFinished(taskId, status);
+ onTaskFinished(task, status);
launchPendingJobs();
}
}
@@ -443,12 +444,12 @@ public class OverlordCompactionScheduler implements
CompactionScheduler
}
}
- private void onTaskFinished(String taskId, TaskStatus taskStatus)
+ private void onTaskFinished(Task task, TaskStatus taskStatus)
{
- statusTracker.onTaskFinished(taskId, taskStatus);
+ statusTracker.onTaskFinished(task.getId(), taskStatus);
updateQueueIfComputed(queue -> {
- queue.onTaskFinished(taskId, taskStatus);
+ queue.onTaskFinished(task.getId(), taskStatus);
updateCompactionSnapshots(queue);
});
}
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/BaseRestorableTaskRunner.java
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/BaseRestorableTaskRunner.java
index be2008ab142..8f888cee1b1 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/BaseRestorableTaskRunner.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/BaseRestorableTaskRunner.java
@@ -152,8 +152,8 @@ public abstract class BaseRestorableTaskRunner<WorkItemType
extends TaskRunnerWo
final Pair<TaskRunnerListener, Executor> listenerPair = Pair.of(listener,
executor);
synchronized (tasks) {
- for (TaskRunnerWorkItem item : tasks.values()) {
- TaskRunnerUtils.notifyLocationChanged(ImmutableList.of(listenerPair),
item.getTaskId(), item.getLocation());
+ for (WorkItemType item : tasks.values()) {
+ notifyLocationChanged(item, listener, executor);
}
listeners.add(listenerPair);
@@ -183,6 +183,12 @@ public abstract class
BaseRestorableTaskRunner<WorkItemType extends TaskRunnerWo
@Override
public abstract RunnerTaskState getRunnerTaskState(String taskId);
+ protected abstract void notifyLocationChanged(
+ WorkItemType item,
+ TaskRunnerListener listener,
+ Executor executor
+ );
+
@Override
public Collection<TaskRunnerWorkItem> getKnownTasks()
{
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/ForkingTaskRunner.java
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/ForkingTaskRunner.java
index 2ec1faf77b8..606f5152063 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/ForkingTaskRunner.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/ForkingTaskRunner.java
@@ -53,6 +53,7 @@ import org.apache.druid.indexing.worker.config.WorkerConfig;
import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.java.util.common.FileUtils;
import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.Pair;
import org.apache.druid.java.util.common.StringUtils;
import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.common.io.Closer;
@@ -87,6 +88,7 @@ import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.Callable;
+import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
@@ -413,10 +415,10 @@ public class ForkingTaskRunner
processHolder.registerWithCloser(closer);
}
- TaskRunnerUtils.notifyLocationChanged(listeners,
task.getId(), taskLocation);
+ TaskRunnerUtils.notifyLocationChanged(listeners, task,
taskLocation);
TaskRunnerUtils.notifyStatusChanged(
listeners,
- task.getId(),
+ task,
TaskStatus.running(task.getId())
);
@@ -444,7 +446,7 @@ public class ForkingTaskRunner
} else {
failedTaskCount.incrementAndGet();
}
- TaskRunnerUtils.notifyStatusChanged(listeners,
task.getId(), status);
+ TaskRunnerUtils.notifyStatusChanged(listeners, task,
status);
return status;
}
catch (Throwable t) {
@@ -697,6 +699,20 @@ public class ForkingTaskRunner
}
}
+ @Override
+ protected void notifyLocationChanged(
+ ForkingTaskRunnerWorkItem item,
+ TaskRunnerListener listener,
+ Executor executor
+ )
+ {
+ TaskRunnerUtils.notifyLocationChanged(
+ List.of(Pair.of(listener, executor)),
+ item.getTask(),
+ item.getLocation()
+ );
+ }
+
@Override
public Optional<ScalingStats> getScalingStats()
{
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/SingleTaskBackgroundRunner.java
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/SingleTaskBackgroundRunner.java
index c16d7212868..361c431c762 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/SingleTaskBackgroundRunner.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/SingleTaskBackgroundRunner.java
@@ -127,7 +127,7 @@ public class SingleTaskBackgroundRunner implements
TaskRunner, QuerySegmentWalke
if (runningItem != null) {
TaskRunnerUtils.notifyLocationChanged(
ImmutableList.of(listenerPair),
- runningItem.getTaskId(),
+ runningItem.getTask(),
runningItem.getLocation()
);
}
@@ -202,7 +202,7 @@ public class SingleTaskBackgroundRunner implements
TaskRunner, QuerySegmentWalke
System.currentTimeMillis() - start
);
- TaskRunnerUtils.notifyStatusChanged(listeners, task.getId(),
taskStatus);
+ TaskRunnerUtils.notifyStatusChanged(listeners, task, taskStatus);
}
catch (Exception e) {
log.makeAlert(e, "Graceful task shutdown failed: %s",
task.getDataSource())
@@ -218,7 +218,7 @@ public class SingleTaskBackgroundRunner implements
TaskRunner, QuerySegmentWalke
// See https://github.com/apache/druid/issues/11445.
TaskRunnerUtils.notifyStatusChanged(
listeners,
- task.getId(),
+ task,
TaskStatus.failure(
task.getId(),
"Failed to stop gracefully with exception. See task logs for
more details."
@@ -233,7 +233,7 @@ public class SingleTaskBackgroundRunner implements
TaskRunner, QuerySegmentWalke
// See https://github.com/apache/druid/issues/11445.
TaskRunnerUtils.notifyStatusChanged(
listeners,
- task.getId(),
+ task,
TaskStatus.failure(task.getId(), "Canceled as task execution
process stopped")
);
}
@@ -471,10 +471,10 @@ public class SingleTaskBackgroundRunner implements
TaskRunner, QuerySegmentWalke
log.info("Running task: %s", task.getId());
TaskRunnerUtils.notifyLocationChanged(
listeners,
- task.getId(),
+ task,
location
);
- TaskRunnerUtils.notifyStatusChanged(listeners, task.getId(),
TaskStatus.running(task.getId()));
+ TaskRunnerUtils.notifyStatusChanged(listeners, task,
TaskStatus.running(task.getId()));
status = task.run(toolbox);
}
catch (InterruptedException e) {
@@ -499,7 +499,7 @@ public class SingleTaskBackgroundRunner implements
TaskRunner, QuerySegmentWalke
}
status = status.withDuration(System.currentTimeMillis() - startTime);
- TaskRunnerUtils.notifyStatusChanged(listeners, task.getId(), status);
+ TaskRunnerUtils.notifyStatusChanged(listeners, task, status);
return status;
}
}
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskRunnerListener.java
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskRunnerListener.java
index 23d11eaa0df..c679848e7b6 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskRunnerListener.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskRunnerListener.java
@@ -21,6 +21,7 @@ package org.apache.druid.indexing.overlord;
import org.apache.druid.indexer.TaskLocation;
import org.apache.druid.indexer.TaskStatus;
+import org.apache.druid.indexing.common.task.Task;
import java.util.concurrent.Executor;
@@ -35,10 +36,10 @@ public interface TaskRunnerListener
* Called when the location of a task has changed. The task may not actually
be done starting up when
* this notification arrives, so it may not be listening at this location
yet.
*/
- void locationChanged(String taskId, TaskLocation newLocation);
+ void locationChanged(Task task, TaskLocation newLocation);
/**
* Called when the status of a task has changed.
*/
- void statusChanged(String taskId, TaskStatus status);
+ void statusChanged(Task task, TaskStatus status);
}
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskRunnerUtils.java
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskRunnerUtils.java
index c2311a63351..99d838aee85 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskRunnerUtils.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskRunnerUtils.java
@@ -26,7 +26,9 @@ import io.netty.handler.codec.http.HttpMethod;
import io.netty.handler.codec.http.HttpResponseStatus;
import org.apache.druid.indexer.TaskLocation;
import org.apache.druid.indexer.TaskStatus;
+import org.apache.druid.indexing.common.task.Task;
import org.apache.druid.indexing.worker.Worker;
+import org.apache.druid.java.util.common.IOE;
import org.apache.druid.java.util.common.Pair;
import org.apache.druid.java.util.common.StringUtils;
import org.apache.druid.java.util.emitter.EmittingLogger;
@@ -51,18 +53,18 @@ public class TaskRunnerUtils
public static void notifyLocationChanged(
final Iterable<Pair<TaskRunnerListener, Executor>> listeners,
- final String taskId,
+ final Task task,
final TaskLocation location
)
{
- log.debug("Task [%s] location changed to [%s].", taskId, location);
+ log.debug("Task [%s] location changed to [%s].", task.getId(), location);
for (final Pair<TaskRunnerListener, Executor> listener : listeners) {
try {
- listener.rhs.execute(() -> listener.lhs.locationChanged(taskId,
location));
+ listener.rhs.execute(() -> listener.lhs.locationChanged(task,
location));
}
catch (Exception e) {
log.makeAlert(e, "Unable to notify task listener")
- .addData("taskId", taskId)
+ .addData("taskId", task.getId())
.addData("taskLocation", location)
.addData("listener", listener.toString())
.emit();
@@ -72,18 +74,18 @@ public class TaskRunnerUtils
public static void notifyStatusChanged(
final Iterable<Pair<TaskRunnerListener, Executor>> listeners,
- final String taskId,
+ final Task task,
final TaskStatus status
)
{
- log.debug("Task [%s] status changed to [%s].", taskId,
status.getStatusCode());
+ log.debug("Task [%s] status changed to [%s].", task.getId(),
status.getStatusCode());
for (final Pair<TaskRunnerListener, Executor> listener : listeners) {
try {
- listener.rhs.execute(() -> listener.lhs.statusChanged(taskId, status));
+ listener.rhs.execute(() -> listener.lhs.statusChanged(task, status));
}
catch (Exception e) {
log.makeAlert(e, "Unable to notify task listener")
- .addData("taskId", taskId)
+ .addData("taskId", task.getId())
.addData("taskStatus", status.getStatusCode())
.addData("listener", listener.toString())
.emit();
@@ -141,8 +143,9 @@ public class TaskRunnerUtils
||
HttpResponseStatus.SERVICE_UNAVAILABLE.equals(responseStatus)) {
return Optional.absent();
} else {
- throw new IOException(
- StringUtils.format("Failed to stream task reports from [%s].
Response status [%s].", url, responseStatus)
+ throw new IOE(
+ "Failed to stream task reports from url[%s]. Response status[%s].",
+ url, responseStatus
);
}
}
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/ThreadingTaskRunner.java
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/ThreadingTaskRunner.java
index 92f577a8f46..999e36a3dd3 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/ThreadingTaskRunner.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/ThreadingTaskRunner.java
@@ -46,6 +46,7 @@ import org.apache.druid.indexing.worker.config.WorkerConfig;
import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.java.util.common.FileUtils;
import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.Pair;
import org.apache.druid.java.util.common.StringUtils;
import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.emitter.EmittingLogger;
@@ -72,6 +73,7 @@ import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.Callable;
+import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
@@ -225,10 +227,10 @@ public class ThreadingTaskRunner
.withTmpStorageBytesPerTask(storageSlot.getNumBytes()),
task
);
- TaskRunnerUtils.notifyLocationChanged(listeners,
task.getId(), taskLocation);
+ TaskRunnerUtils.notifyLocationChanged(listeners,
task, taskLocation);
TaskRunnerUtils.notifyStatusChanged(
listeners,
- task.getId(),
+ task,
TaskStatus.running(task.getId())
);
@@ -264,7 +266,7 @@ public class ThreadingTaskRunner
Appenderators.clearTaskThreadContextForIndexers();
}
- TaskRunnerUtils.notifyStatusChanged(listeners,
task.getId(), taskStatus);
+ TaskRunnerUtils.notifyStatusChanged(listeners,
task, taskStatus);
return taskStatus;
}
catch (Throwable t) {
@@ -476,6 +478,20 @@ public class ThreadingTaskRunner
return workItem == null ? null : workItem.getState();
}
+ @Override
+ protected void notifyLocationChanged(
+ ThreadingTaskRunnerWorkItem item,
+ TaskRunnerListener listener,
+ Executor executor
+ )
+ {
+ TaskRunnerUtils.notifyLocationChanged(
+ List.of(Pair.of(listener, executor)),
+ item.getTask(),
+ item.getLocation()
+ );
+ }
+
private Collection<TaskRunnerWorkItem> getTasks(RunnerTaskState state)
{
synchronized (tasks) {
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 b237b6577ae..24f0fe12752 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
@@ -413,7 +413,7 @@ public class HttpRemoteTaskRunner implements
WorkerTaskRunner, TaskLogStreamer,
} else {
// Notify interested parties
taskRunnerWorkItem.setResult(taskStatus);
- TaskRunnerUtils.notifyStatusChanged(listeners, taskStatus.getId(),
taskStatus);
+ TaskRunnerUtils.notifyStatusChanged(listeners, taskRunnerWorkItem.task,
taskStatus);
// Update success/failure counters, Blacklist node if there are too many
failures.
if (workerHolder != null) {
@@ -1004,7 +1004,7 @@ public class HttpRemoteTaskRunner implements
WorkerTaskRunner, TaskLogStreamer,
if (entry.getValue().getState() ==
HttpRemoteTaskRunnerWorkItem.State.RUNNING) {
TaskRunnerUtils.notifyLocationChanged(
ImmutableList.of(listenerPair),
- entry.getKey(),
+ entry.getValue().getTask(),
entry.getValue().getLocation()
);
}
@@ -1442,11 +1442,21 @@ public class HttpRemoteTaskRunner implements
WorkerTaskRunner, TaskLogStreamer,
if (knownStatusInStorage.isPresent()) {
switch (knownStatusInStorage.get().getStatusCode()) {
case RUNNING:
+ final Optional<Task> task = taskStorage.getTask(taskId);
+ if (!task.isPresent()) {
+ log.makeAlert(
+ "Could not fetch payload of task[%s] with status[%s]."
+ + " Ignoring notification[%s] from worker[%s].",
+ taskId, knownStatusInStorage.get(), announcement,
worker.getHost()
+ ).emit();
+ break;
+ }
+
taskItem = new HttpRemoteTaskRunnerWorkItem(
taskId,
worker,
TaskLocation.unknown(),
- null,
+ task.get(),
announcement.getTaskType(),
HttpRemoteTaskRunnerWorkItem.State.RUNNING
);
@@ -1516,7 +1526,7 @@ public class HttpRemoteTaskRunner implements
WorkerTaskRunner, TaskLogStreamer,
announcement.getTaskLocation()
);
taskItem.setLocation(announcement.getTaskLocation());
- TaskRunnerUtils.notifyLocationChanged(listeners, taskId,
announcement.getTaskLocation());
+ TaskRunnerUtils.notifyLocationChanged(listeners,
taskItem.getTask(), announcement.getTaskLocation());
}
} else {
log.warn(
@@ -1565,7 +1575,7 @@ public class HttpRemoteTaskRunner implements
WorkerTaskRunner, TaskLogStreamer,
announcement.getTaskLocation()
);
taskItem.setLocation(announcement.getTaskLocation());
- TaskRunnerUtils.notifyLocationChanged(listeners, taskId,
announcement.getTaskLocation());
+ TaskRunnerUtils.notifyLocationChanged(listeners,
taskItem.getTask(), announcement.getTaskLocation());
}
isTaskCompleted = true;
@@ -1790,17 +1800,14 @@ public class HttpRemoteTaskRunner implements
WorkerTaskRunner, TaskLogStreamer,
String taskId,
Worker worker,
TaskLocation location,
- @Nullable Task task,
+ Task task,
String taskType,
State state
)
{
- super(taskId, task == null ? null : task.getType(), worker, location,
task == null ? null : task.getDataSource());
+ super(taskId, task.getType(), worker, location, task.getDataSource());
this.state = Preconditions.checkNotNull(state);
- Preconditions.checkArgument(task == null || taskType == null ||
taskType.equals(task.getType()));
-
- // It is possible to have it null when the TaskRunner is just started
and discovered this taskId from a worker,
- // notifications don't contain whole Task instance but just metadata
about the task.
+ Preconditions.checkArgument(taskType == null ||
taskType.equals(task.getType()));
this.task = task;
}
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/scheduledbatch/ScheduledBatchTaskManager.java
b/indexing-service/src/main/java/org/apache/druid/indexing/scheduledbatch/ScheduledBatchTaskManager.java
index 860710220e2..64b9d5f978d 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/scheduledbatch/ScheduledBatchTaskManager.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/scheduledbatch/ScheduledBatchTaskManager.java
@@ -25,6 +25,7 @@ import org.apache.druid.client.broker.BrokerClient;
import org.apache.druid.common.guava.FutureUtils;
import org.apache.druid.indexer.TaskLocation;
import org.apache.druid.indexer.TaskStatus;
+import org.apache.druid.indexing.common.task.Task;
import org.apache.druid.indexing.overlord.LeaderOverlordService;
import org.apache.druid.indexing.overlord.TaskMaster;
import org.apache.druid.indexing.overlord.TaskRunner;
@@ -103,16 +104,16 @@ public class ScheduledBatchTaskManager implements
LeaderOverlordService
}
@Override
- public void locationChanged(String taskId, TaskLocation newLocation)
+ public void locationChanged(Task task, TaskLocation newLocation)
{
// Do nothing
}
@Override
- public void statusChanged(final String taskId, final TaskStatus
taskStatus)
+ public void statusChanged(final Task task, final TaskStatus taskStatus)
{
if (taskStatus.isComplete()) {
- statusTracker.onTaskCompleted(taskId, taskStatus);
+ statusTracker.onTaskCompleted(task.getId(), taskStatus);
}
}
};
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java
index ab204a63acf..3377090a29c 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java
@@ -2099,13 +2099,13 @@ public abstract class
SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
}
@Override
- public void locationChanged(final String taskId, final
TaskLocation newLocation)
+ public void locationChanged(final Task task, final TaskLocation
newLocation)
{
// do nothing
}
@Override
- public void statusChanged(String taskId, TaskStatus status)
+ public void statusChanged(Task task, TaskStatus status)
{
addNotice(new RunNotice());
}
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/worker/WorkerTaskManager.java
b/indexing-service/src/main/java/org/apache/druid/indexing/worker/WorkerTaskManager.java
index 996831badba..cea63c937ac 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/worker/WorkerTaskManager.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/worker/WorkerTaskManager.java
@@ -240,13 +240,13 @@ public class WorkerTaskManager implements
IndexerTaskCountStatsProvider
}
@Override
- public void locationChanged(final String taskId, final TaskLocation
newLocation)
+ public void locationChanged(final Task task, final TaskLocation
newLocation)
{
- submitNoticeToExec(new LocationNotice(taskId, newLocation));
+ submitNoticeToExec(new LocationNotice(task.getId(), newLocation));
}
@Override
- public void statusChanged(final String taskId, final TaskStatus
status)
+ public void statusChanged(final Task task, final TaskStatus status)
{
// do nothing
}
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/SingleTaskBackgroundRunnerTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/SingleTaskBackgroundRunnerTest.java
index 1845c38db52..0a73e82ad31 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/SingleTaskBackgroundRunnerTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/SingleTaskBackgroundRunnerTest.java
@@ -35,6 +35,7 @@ import org.apache.druid.indexing.common.config.TaskConfig;
import org.apache.druid.indexing.common.config.TaskConfigBuilder;
import org.apache.druid.indexing.common.task.AbstractTask;
import org.apache.druid.indexing.common.task.NoopTask;
+import org.apache.druid.indexing.common.task.Task;
import org.apache.druid.indexing.common.task.TestAppenderatorsManager;
import org.apache.druid.java.util.common.Intervals;
import org.apache.druid.java.util.common.concurrent.Execs;
@@ -236,13 +237,13 @@ public class SingleTaskBackgroundRunnerTest
}
@Override
- public void locationChanged(String taskId, TaskLocation newLocation)
+ public void locationChanged(Task task, TaskLocation newLocation)
{
// do nothing
}
@Override
- public void statusChanged(String taskId, TaskStatus status)
+ public void statusChanged(Task task, TaskStatus status)
{
statusHolder.set(status);
}
@@ -299,13 +300,13 @@ public class SingleTaskBackgroundRunnerTest
}
@Override
- public void locationChanged(String taskId, TaskLocation newLocation)
+ public void locationChanged(Task task, TaskLocation newLocation)
{
// do nothing
}
@Override
- public void statusChanged(String taskId, TaskStatus status)
+ public void statusChanged(Task task, TaskStatus status)
{
if (status.getStatusCode() == TaskState.RUNNING) {
runLatch.countDown();
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TestTaskRunner.java
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TestTaskRunner.java
index 2afb2b4dae3..01523c21166 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TestTaskRunner.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TestTaskRunner.java
@@ -111,7 +111,7 @@ public class TestTaskRunner implements TaskRunner,
QuerySegmentWalker
listeners.add(listenerPair);
log.info("Registered listener [%s]", listener.getListenerId());
for (TestTaskRunnerWorkItem item : runningItems) {
- TaskRunnerUtils.notifyLocationChanged(ImmutableList.of(listenerPair),
item.getTaskId(), item.getLocation());
+ TaskRunnerUtils.notifyLocationChanged(ImmutableList.of(listenerPair),
item.getTask(), item.getLocation());
}
}
@@ -173,15 +173,15 @@ public class TestTaskRunner implements TaskRunner,
QuerySegmentWalker
System.currentTimeMillis() - start
);
- TaskRunnerUtils.notifyStatusChanged(listeners, task.getId(),
taskStatus);
+ TaskRunnerUtils.notifyStatusChanged(listeners, task, taskStatus);
}
catch (Exception e) {
String errMsg = "Graceful shutdown of task aborted with exception,
see task logs for more information";
- TaskRunnerUtils.notifyStatusChanged(listeners, task.getId(),
TaskStatus.failure(task.getId(), errMsg));
+ TaskRunnerUtils.notifyStatusChanged(listeners, task,
TaskStatus.failure(task.getId(), errMsg));
throw new RE(e, "Graceful shutdown of task[%s] aborted with
exception", task.getId());
}
} else {
- TaskRunnerUtils.notifyStatusChanged(listeners, task.getId(),
TaskStatus.failure(
+ TaskRunnerUtils.notifyStatusChanged(listeners, task,
TaskStatus.failure(
task.getId(),
"Task failure while shutting down gracefully"
));
@@ -405,10 +405,10 @@ public class TestTaskRunner implements TaskRunner,
QuerySegmentWalker
log.info("Running task: %s", task.getId());
TaskRunnerUtils.notifyLocationChanged(
listeners,
- task.getId(),
+ task,
taskLocation
);
- TaskRunnerUtils.notifyStatusChanged(listeners, task.getId(),
TaskStatus.running(task.getId()));
+ TaskRunnerUtils.notifyStatusChanged(listeners, task,
TaskStatus.running(task.getId()));
status = task.run(toolbox);
}
catch (InterruptedException e) {
@@ -432,7 +432,7 @@ public class TestTaskRunner implements TaskRunner,
QuerySegmentWalker
}
status = status.withDuration(System.currentTimeMillis() - startTime);
- TaskRunnerUtils.notifyStatusChanged(listeners, task.getId(), status);
+ TaskRunnerUtils.notifyStatusChanged(listeners, task, status);
return status;
}
}
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 8a053ecc4a5..6d965b529b9 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
@@ -62,10 +62,12 @@ import org.apache.druid.java.util.common.ISE;
import org.apache.druid.java.util.common.StringUtils;
import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.emitter.EmittingLogger;
+import org.apache.druid.java.util.emitter.service.AlertEvent;
import org.apache.druid.java.util.http.client.HttpClient;
import org.apache.druid.java.util.http.client.Request;
import
org.apache.druid.java.util.http.client.response.InputStreamFullResponseHandler;
import
org.apache.druid.java.util.http.client.response.InputStreamFullResponseHolder;
+import org.apache.druid.java.util.metrics.StubServiceEmitter;
import org.apache.druid.segment.TestHelper;
import org.apache.druid.server.DruidNode;
import org.apache.druid.server.coordination.ChangeRequestHttpSyncer;
@@ -101,10 +103,13 @@ import static org.easymock.EasyMock.isA;
*/
public class HttpRemoteTaskRunnerTest
{
+ private StubServiceEmitter emitter;
+
@BeforeEach
public void setup()
{
- EmittingLogger.registerEmitter(new NoopServiceEmitter());
+ emitter = new StubServiceEmitter();
+ EmittingLogger.registerEmitter(emitter);
}
/*
@@ -451,9 +456,15 @@ public class HttpRemoteTaskRunnerTest
TaskStorage taskStorageMock = EasyMock.createStrictMock(TaskStorage.class);
EasyMock.expect(taskStorageMock.getStatus(task1.getId())).andReturn(Optional.absent());
EasyMock.expect(taskStorageMock.getStatus(task2.getId())).andReturn(Optional.absent()).times(2);
+
EasyMock.expect(taskStorageMock.getStatus(task3.getId())).andReturn(Optional.of(TaskStatus.running(task3.getId())));
+
EasyMock.expect(taskStorageMock.getTask(task3.getId())).andReturn(Optional.of(task3));
+
EasyMock.expect(taskStorageMock.getStatus(task4.getId())).andReturn(Optional.of(TaskStatus.running(task4.getId())));
+
EasyMock.expect(taskStorageMock.getTask(task4.getId())).andReturn(Optional.of(task4));
+
EasyMock.expect(taskStorageMock.getStatus(task5.getId())).andReturn(Optional.of(TaskStatus.success(task5.getId())));
+
EasyMock.replay(taskStorageMock);
HttpRemoteTaskRunner taskRunner = new HttpRemoteTaskRunner(
@@ -1396,7 +1407,9 @@ public class HttpRemoteTaskRunnerTest
TaskStorage taskStorage = EasyMock.createMock(TaskStorage.class);
EasyMock.expect(taskStorage.getStatus(task1.getId())).andReturn(Optional.of(TaskStatus.running(task1.getId())));
+
EasyMock.expect(taskStorage.getTask(task1.getId())).andReturn(Optional.of(task1));
EasyMock.expect(taskStorage.getStatus(task2.getId())).andReturn(Optional.of(TaskStatus.running(task2.getId())));
+
EasyMock.expect(taskStorage.getTask(task2.getId())).andReturn(Optional.of(task2));
EasyMock.expect(taskStorage.getStatus(task3.getId())).andReturn(Optional.of(TaskStatus.success(task3.getId())));
EasyMock.expect(taskStorage.getStatus(task4.getId())).andReturn(Optional.of(TaskStatus.success(task4.getId())));
EasyMock.expect(taskStorage.getStatus(task5.getId())).andReturn(Optional.absent());
@@ -1468,6 +1481,54 @@ public class HttpRemoteTaskRunnerTest
);
}
+ @Test
+ public void
test_taskAddedOrUpdated_taskNotFoundInStorage_raisesAlert_andShutsDownTask()
+ {
+ final Task task = NoopTask.create();
+
+ // Setup storage to return a task status but not the task payload
+ final TaskStatus statusInStorage = TaskStatus.running(task.getId());
+
+ final TaskStorage taskStorage = EasyMock.createMock(TaskStorage.class);
+
EasyMock.expect(taskStorage.getStatus(task.getId())).andReturn(Optional.of(statusInStorage));
+
EasyMock.expect(taskStorage.getTask(task.getId())).andReturn(Optional.absent());
+ EasyMock.replay(taskStorage);
+
+ final HttpRemoteTaskRunner taskRunner =
+ createTaskRunnerForTestTaskAddedOrUpdated(taskStorage, new
ArrayList<>());
+
+ // Expect worker to get shutdown notification
+ final Worker worker = new Worker("http", "localhost", "127.0.0.1", 1,
"v1", WorkerConfig.DEFAULT_CATEGORY);
+ final WorkerHolder workerHolder = EasyMock.createMock(WorkerHolder.class);
+ EasyMock.expect(workerHolder.getWorker()).andReturn(worker).anyTimes();
+ workerHolder.shutdownTask(task.getId());
+ EasyMock.replay(workerHolder);
+
+ Assertions.assertEquals(0, taskRunner.getKnownTasks().size());
+
+ // Send the announcement to the TaskRunner
+ final TaskAnnouncement announcement = TaskAnnouncement.create(
+ task,
+ TaskStatus.running(task.getId()),
+ TaskLocation.create("worker", 1, 2)
+ );
+ taskRunner.taskAddedOrUpdated(announcement, workerHolder);
+
+ // Verify that alert has been raised
+ final List<AlertEvent> alerts = emitter.getAlerts();
+ Assertions.assertEquals(1, alerts.size());
+ Assertions.assertEquals(
+ StringUtils.format(
+ "Could not fetch payload of task[%s] with status[%s]."
+ + " Ignoring notification[%s] from worker[%s].",
+ task.getId(), statusInStorage, announcement, worker.getHost()
+ ),
+ alerts.getFirst().getDescription()
+ );
+
+ EasyMock.verify(workerHolder, taskStorage);
+ }
+
@Test
public void testTimeoutInAssigningTasks() throws Exception
{
@@ -1905,15 +1966,15 @@ public class HttpRemoteTaskRunnerTest
}
@Override
- public void locationChanged(String taskId, TaskLocation
newLocation)
+ public void locationChanged(Task task, TaskLocation newLocation)
{
- listenerNotificationsAccumulator.add(ImmutableList.of(taskId,
newLocation));
+
listenerNotificationsAccumulator.add(ImmutableList.of(task.getId(),
newLocation));
}
@Override
- public void statusChanged(String taskId, TaskStatus status)
+ public void statusChanged(Task task, TaskStatus status)
{
- listenerNotificationsAccumulator.add(ImmutableList.of(taskId,
status));
+
listenerNotificationsAccumulator.add(ImmutableList.of(task.getId(), status));
}
},
Execs.directExecutor()
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
