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 83e6caf41ea fix: Keep task count dimensions for tasks the task runner
still reports (#20735)
83e6caf41ea is described below
commit 83e6caf41ea087bf81294560eb92d514c15e953e
Author: Andreas Maechler <[email protected]>
AuthorDate: Mon Nov 2 21:26:03 2026 -0600
fix: Keep task count dimensions for tasks the task runner still reports
(#20735)
* Fall back to work item dimensions in running and pending task counts
TaskQueue keyed runner-reported tasks without a non-complete entry in
activeTasks as RowKey.empty(), so task/running/count and
task/pending/count emitted them without dataSource or taskType. This
happens while a task killed through TaskQueue.shutdown() is still
stopping on the runner, and for runner tasks the queue no longer
tracks. Build the key from the work item's datasource and task type
instead, keeping the empty key only when either is null.
* Keep full metric keys for completed tasks still on the runner
Look up running and pending tasks in all activeTasks entries, including
completed ones, so a task shut down through TaskQueue.shutdown() keeps
its full metric key, including supervisorId for streaming tasks, until
the next storage sync removes its entry. The work-item fallback now only
applies to tasks with no entry left in activeTasks. The work item has no
supervisorId, so those rows stay without it.
The fallback also keeps whichever of datasource and task type is
non-null, instead of dropping both when one is missing.
* Look up runner tasks in activeTasks directly for task counts
Running and pending task counts built a map of every active task's
metric key on each call, only to look up the tasks the runner reports.
Look each runner task up in activeTasks instead, share one counting
method between both counts, and remove getCurrentTaskDatasources().
* Share work item construction in TaskQueueTest
* Cover running tasks removed from the queue in TaskQueueTest
Add a streaming task that is shut down and removed by
syncFromStorage() while the runner still reports it, which falls back
to the work item and has no supervisorId. Rename the test to match.
* Use the work item's task for task count keys after queue removal
Add a nullable getTask() to TaskRunnerWorkItem, returning null by
default. The five in-tree work items already had this method and now
override it. getWorkItemMetricKey uses it for tasks no longer in
activeTasks, so a streaming task the runner still reports keeps its
supervisorId and merges into its supervisor's row.
Without it, that row had only dataSource and taskType. The StatsD
emitter's default config emits task/running/count as a gauge keyed by
those two dimensions, so the row overwrote the supervisor's main row.
The datasource and task type fallback remains for work items that do
not expose the task.
* Drop duplicate removed stream task case from TaskQueueTest
* Drop the datasource and task type fallback for task count keys
Every in-tree work item now exposes its task, so the fallback to the
work item's datasource and task type only ran for third-party work
items. For a removed streaming task it emitted a row without
supervisorId next to the supervisor's main row, which the StatsD
default config overwrites. Return the empty key when neither
activeTasks nor the work item has the task, as before this change.
* Count task work items with groupingBy instead of toMap
---
.../druid/k8s/overlord/KubernetesWorkItem.java | 1 +
.../druid/indexing/overlord/ForkingTaskRunner.java | 1 +
.../overlord/SingleTaskBackgroundRunner.java | 1 +
.../apache/druid/indexing/overlord/TaskQueue.java | 38 ++++-----
.../indexing/overlord/TaskRunnerWorkItem.java | 10 +++
.../indexing/overlord/ThreadingTaskRunner.java | 1 +
.../overlord/hrtr/HttpRemoteTaskRunner.java | 1 +
.../druid/indexing/overlord/TaskQueueTest.java | 98 ++++++++++++++++++++++
.../druid/indexing/overlord/TestTaskRunner.java | 1 +
9 files changed, 129 insertions(+), 23 deletions(-)
diff --git
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesWorkItem.java
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesWorkItem.java
index bcc268805e4..e6ef276cf72 100644
---
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesWorkItem.java
+++
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesWorkItem.java
@@ -115,6 +115,7 @@ public class KubernetesWorkItem extends TaskRunnerWorkItem
return task.getDataSource();
}
+ @Override
public Task getTask()
{
return task;
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 01b480a0516..078f4b9d25f 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
@@ -891,6 +891,7 @@ public class ForkingTaskRunner
this.task = task;
}
+ @Override
public Task getTask()
{
return task;
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 8e0ed1c33cb..36557fb3b0e 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
@@ -423,6 +423,7 @@ public class SingleTaskBackgroundRunner implements
TaskRunner, QuerySegmentWalke
this.location = location;
}
+ @Override
public Task getTask()
{
return task;
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskQueue.java
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskQueue.java
index 405e6f8f60d..366810ec9dc 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskQueue.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskQueue.java
@@ -927,36 +927,19 @@ public class TaskQueue
}
}
- private Map<String, RowKey> getCurrentTaskDatasources()
+ public Map<RowKey, Long> getRunningTaskCount()
{
- return activeTasks.values().stream()
- .filter(entry -> !entry.isComplete)
- .map(TaskEntry::getTask)
- .collect(Collectors.toMap(Task::getId,
TaskQueue::getMetricKey));
+ return getTaskCountByMetricKey(taskRunner.getRunningTasks());
}
- public Map<RowKey, Long> getRunningTaskCount()
+ public Map<RowKey, Long> getPendingTaskCount()
{
- final Map<String, RowKey> taskDatasources = getCurrentTaskDatasources();
- return taskRunner.getRunningTasks()
- .stream()
- .collect(Collectors.toMap(
- e -> taskDatasources.getOrDefault(e.getTaskId(),
RowKey.empty()),
- e -> 1L,
- Long::sum
- ));
+ return getTaskCountByMetricKey(taskRunner.getPendingTasks());
}
- public Map<RowKey, Long> getPendingTaskCount()
+ private Map<RowKey, Long> getTaskCountByMetricKey(final Collection<? extends
TaskRunnerWorkItem> workItems)
{
- final Map<String, RowKey> taskDatasources = getCurrentTaskDatasources();
- return taskRunner.getPendingTasks()
- .stream()
- .collect(Collectors.toMap(
- e -> taskDatasources.getOrDefault(e.getTaskId(),
RowKey.empty()),
- e -> 1L,
- Long::sum
- ));
+ return
workItems.stream().collect(Collectors.groupingBy(this::getWorkItemMetricKey,
Collectors.counting()));
}
public Map<RowKey, Long> getWaitingTaskCount()
@@ -1236,4 +1219,13 @@ public class TaskQueue
}
return builder.build();
}
+
+ /**
+ * Returns the key of the task in {@link #activeTasks} or on the work item,
or an empty key if neither has it.
+ */
+ private RowKey getWorkItemMetricKey(final TaskRunnerWorkItem workItem)
+ {
+ final TaskEntry entry = activeTasks.get(workItem.getTaskId());
+ return getMetricKey(entry != null ? entry.getTask() : workItem.getTask());
+ }
}
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskRunnerWorkItem.java
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskRunnerWorkItem.java
index f9aa6652d23..b8a6520f169 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskRunnerWorkItem.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskRunnerWorkItem.java
@@ -24,6 +24,7 @@ import com.fasterxml.jackson.annotation.JsonProperty;
import com.google.common.util.concurrent.ListenableFuture;
import org.apache.druid.indexer.TaskLocation;
import org.apache.druid.indexer.TaskStatus;
+import org.apache.druid.indexing.common.task.Task;
import org.apache.druid.java.util.common.DateTimes;
import org.joda.time.DateTime;
@@ -96,6 +97,15 @@ public abstract class TaskRunnerWorkItem
public abstract String getTaskType();
public abstract String getDataSource();
+ /**
+ * Returns the task of this work item, or null if this runner does not keep
it.
+ */
+ @Nullable
+ public Task getTask()
+ {
+ return null;
+ }
+
@Override
public String toString()
{
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 b057e0409ef..a5b17f7b119 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
@@ -587,6 +587,7 @@ public class ThreadingTaskRunner
this.state = RunnerTaskState.PENDING;
}
+ @Override
public Task getTask()
{
return task;
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 2378c737763..7258f0357a9 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
@@ -1832,6 +1832,7 @@ public class HttpRemoteTaskRunner implements
WorkerTaskRunner, TaskLogStreamer,
Objects.equal(getWorker().getHost(), candidateWorker.getHost());
}
+ @Override
public Task getTask()
{
return task;
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskQueueTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskQueueTest.java
index 531665027cf..f66aeaabc37 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskQueueTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskQueueTest.java
@@ -68,6 +68,9 @@ import
org.apache.druid.indexing.overlord.hrtr.HttpRemoteTaskRunner;
import org.apache.druid.indexing.overlord.hrtr.HttpRemoteTaskRunnerTest;
import org.apache.druid.indexing.overlord.hrtr.WorkerHolder;
import org.apache.druid.indexing.overlord.setup.DefaultWorkerBehaviorConfig;
+import org.apache.druid.indexing.seekablestream.TestSeekableStreamIndexTask;
+import
org.apache.druid.indexing.seekablestream.TestSeekableStreamIndexTaskIOConfig;
+import
org.apache.druid.indexing.seekablestream.supervisor.SeekableStreamSupervisorStateTest;
import org.apache.druid.indexing.test.TestIndexerMetadataStorageCoordinator;
import org.apache.druid.indexing.worker.TaskAnnouncement;
import org.apache.druid.indexing.worker.Worker;
@@ -87,6 +90,8 @@ import org.apache.druid.query.DruidMetrics;
import org.apache.druid.segment.indexing.DataSchema;
import org.apache.druid.server.DruidNode;
import org.apache.druid.server.coordinator.stats.CoordinatorRunStats;
+import org.apache.druid.server.coordinator.stats.Dimension;
+import org.apache.druid.server.coordinator.stats.RowKey;
import org.apache.druid.timeline.DataSegment;
import org.easymock.EasyMock;
import org.joda.time.Interval;
@@ -100,6 +105,7 @@ import java.io.IOException;
import java.net.URI;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@@ -803,6 +809,62 @@ public class TaskQueueTest extends IngestionTestBase
Assertions.assertEquals(lowPriority1.getId(), submitted.get(4));
}
+ @Test
+ public void testRunningAndPendingTaskCountForCompletedAndRemovedTasks()
+ {
+ final TestTask activeTask = new TestTask("active",
Intervals.of("2021-01/P1M"));
+ final TestTask completedTask = new TestTask("completed",
Intervals.of("2021-02/P1M"));
+ final TestSeekableStreamIndexTask completedStreamTask =
createStreamTask("completedStream");
+ final TestSeekableStreamIndexTask removedStreamTask =
createStreamTask("removedStream");
+
+ final TaskRunner taskRunner = EasyMock.createNiceMock(TaskRunner.class);
+ EasyMock.<Collection<? extends
TaskRunnerWorkItem>>expect(taskRunner.getRunningTasks()).andReturn(List.of(
+ workItem(activeTask),
+ workItem(completedTask),
+ workItem(completedStreamTask),
+ workItemWithTask(removedStreamTask)
+ )).anyTimes();
+ EasyMock.<Collection<? extends
TaskRunnerWorkItem>>expect(taskRunner.getPendingTasks()).andReturn(List.of(
+ new RemoteTaskRunnerWorkItem("unknown", "index_kafka", null, null,
"wiki")
+ )).anyTimes();
+ EasyMock.replay(taskRunner);
+
+ final TaskQueue queue = new TaskQueue(
+ new TaskLockConfig(),
+ new TaskQueueConfig(null, null, null, null, null, null),
+ new DefaultTaskConfig(),
+ getTaskStorage(),
+ taskRunner,
+ actionClientFactory,
+ getLockbox(),
+ serviceEmitter,
+ getObjectMapper(),
+ new NoopTaskContextEnricher()
+ );
+ queue.setActive(true);
+ queue.add(removedStreamTask);
+ queue.shutdown(removedStreamTask.getId(), "test shutdown");
+ queue.syncFromStorage();
+
Assertions.assertFalse(queue.getActiveTaskInfo(removedStreamTask.getId()).isPresent());
+
+ queue.add(activeTask);
+ queue.add(completedTask);
+ queue.add(completedStreamTask);
+ queue.shutdown(completedTask.getId(), "test shutdown");
+ queue.shutdown(completedStreamTask.getId(), "test shutdown");
+
+ Assertions.assertEquals(
+ Map.of(
+ RowKey.with(Dimension.DATASOURCE,
"datasource").and(Dimension.TASK_TYPE, "test"), 2L,
+ RowKey.with(Dimension.DATASOURCE, "stream")
+ .with(Dimension.TASK_TYPE, "test_seekable_stream")
+ .and(Dimension.SUPERVISOR_ID, "supervisor"), 2L
+ ),
+ queue.getRunningTaskCount()
+ );
+ Assertions.assertEquals(Map.of(RowKey.empty(), 1L),
queue.getPendingTaskCount());
+ }
+
private HttpRemoteTaskRunner createHttpRemoteTaskRunner()
{
final DruidNodeDiscoveryProvider druidNodeDiscoveryProvider
@@ -851,6 +913,42 @@ public class TaskQueueTest extends IngestionTestBase
);
}
+ private static TaskRunnerWorkItem workItem(final Task task)
+ {
+ return new RemoteTaskRunnerWorkItem(task.getId(), task.getType(), null,
null, task.getDataSource());
+ }
+
+ private static TaskRunnerWorkItem workItemWithTask(final Task task)
+ {
+ return new RemoteTaskRunnerWorkItem(task.getId(), task.getType(), null,
null, task.getDataSource())
+ {
+ @Override
+ public Task getTask()
+ {
+ return task;
+ }
+ };
+ }
+
+ private static TestSeekableStreamIndexTask createStreamTask(final String id)
+ {
+ return new TestSeekableStreamIndexTask(
+ id,
+ "supervisor",
+ null,
+ DataSchema.builder()
+ .withDataSource("stream")
+ .withTimestamp(TimestampSpec.DEFAULT)
+ .withDimensions(DimensionsSpec.builder().build())
+ .withGranularity(new
UniformGranularitySpec(Granularities.HOUR, Granularities.NONE, null))
+ .build(),
+
SeekableStreamSupervisorStateTest.createSupervisorTuningConfig().convertToTaskTuningConfig(),
+ new TestSeekableStreamIndexTaskIOConfig(),
+ null,
+ null
+ );
+ }
+
private static class TestTask extends AbstractBatchIndexTask
{
private final Interval interval;
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 01523c21166..fa77a633337 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
@@ -332,6 +332,7 @@ public class TestTaskRunner implements TaskRunner,
QuerySegmentWalker
this.task = task;
}
+ @Override
public Task getTask()
{
return task;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
