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 9733020550c fix: Stop task count metrics from leaking dimensions
across rows (#20726)
9733020550c is described below
commit 9733020550c22999e00f0d1920f634e5f6b1ba02
Author: Andreas Maechler <[email protected]>
AuthorDate: Mon Oct 26 10:01:31 2026 -0600
fix: Stop task count metrics from leaking dimensions across rows (#20726)
* Fix dimensions leaking across rows in TaskCountStatsMonitor
TaskCountStatsMonitor reused one ServiceMetricEvent.Builder for every row
of a
task count metric. Builder dimensions accumulate, so a row without
supervisorId
inherited the previous row's value, and an empty row key inherited the
previous
row's dataSource and taskType. Create a new builder per row.
* Simplify TaskCountStatsMonitor regression test setup
---
.../server/metrics/TaskCountStatsMonitor.java | 2 +-
.../server/metrics/TaskCountStatsMonitorTest.java | 30 ++++++++++++++++++++++
2 files changed, 31 insertions(+), 1 deletion(-)
diff --git
a/server/src/main/java/org/apache/druid/server/metrics/TaskCountStatsMonitor.java
b/server/src/main/java/org/apache/druid/server/metrics/TaskCountStatsMonitor.java
index f6598c0e115..beafbb4d62b 100644
---
a/server/src/main/java/org/apache/druid/server/metrics/TaskCountStatsMonitor.java
+++
b/server/src/main/java/org/apache/druid/server/metrics/TaskCountStatsMonitor.java
@@ -67,9 +67,9 @@ public class TaskCountStatsMonitor extends AbstractMonitor
private void emit(ServiceEmitter emitter, String key, Map<RowKey, Long>
counts)
{
- final ServiceMetricEvent.Builder builder = new
ServiceMetricEvent.Builder();
if (counts != null) {
counts.forEach((k, v) -> {
+ final ServiceMetricEvent.Builder builder = new
ServiceMetricEvent.Builder();
k.getValues().forEach((dim, value) ->
builder.setDimension(dim.reportedName(), value));
emitter.emit(builder.setMetric(key, v));
});
diff --git
a/server/src/test/java/org/apache/druid/server/metrics/TaskCountStatsMonitorTest.java
b/server/src/test/java/org/apache/druid/server/metrics/TaskCountStatsMonitorTest.java
index 03b8cef7007..33600b2bd60 100644
---
a/server/src/test/java/org/apache/druid/server/metrics/TaskCountStatsMonitorTest.java
+++
b/server/src/test/java/org/apache/druid/server/metrics/TaskCountStatsMonitorTest.java
@@ -20,6 +20,7 @@
package org.apache.druid.server.metrics;
import com.google.common.collect.ImmutableMap;
+import org.apache.druid.java.util.emitter.service.ServiceMetricEvent;
import org.apache.druid.java.util.metrics.StubServiceEmitter;
import org.apache.druid.server.coordinator.stats.CoordinatorRunStats;
import org.apache.druid.server.coordinator.stats.CoordinatorStat;
@@ -28,7 +29,9 @@ import org.apache.druid.server.coordinator.stats.RowKey;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+import java.util.List;
import java.util.Map;
public class TaskCountStatsMonitorTest
@@ -108,6 +111,33 @@ public class TaskCountStatsMonitorTest
emitter.verifyValue(Stat.DEBUG_1.getMetricName(), Map.of("tier", "hot",
"dataSource", "wiki"), 20L);
}
+ @Test
+ public void testDimensionsDoNotLeakAcrossRows()
+ {
+ final RowKey streamingKey = RowKey.with(Dimension.DATASOURCE, "d1")
+ .with(Dimension.TASK_TYPE, "index_kafka")
+ .with(Dimension.SUPERVISOR_ID, "s1")
+ .build();
+ final TaskCountStatsProvider provider =
Mockito.mock(TaskCountStatsProvider.class);
+ Mockito.when(provider.getRunningTaskCount())
+ .thenReturn(ImmutableMap.of(streamingKey, 3L, TASK_METRIC_KEY2, 1L,
RowKey.empty(), 1L));
+
+ final StubServiceEmitter emitter = new StubServiceEmitter("service",
"host");
+ new TaskCountStatsMonitor(provider).doMonitor(emitter);
+
+ Assertions.assertEquals(
+ List.of(
+ Map.of("dataSource", "d1", "taskType", "index_kafka",
"supervisorId", "s1"),
+ Map.of("dataSource", "d1", "taskType", "kill"),
+ Map.of()
+ ),
+ emitter.getMetricEvents("task/running/count")
+ .stream()
+ .map(ServiceMetricEvent::getUserDims)
+ .toList()
+ );
+ }
+
private static class Stat
{
static final CoordinatorStat INFO_1 = CoordinatorStat.toLogAndEmit("i1",
"info/1", CoordinatorStat.Level.INFO);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
