This is an automated email from the ASF dual-hosted git repository.
cryptoe 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 c04fd0a532d fix: do not report poll-idle-ratio as 1.0 when it is
unknown (#20783)
c04fd0a532d is described below
commit c04fd0a532d3488cd536f3697151a7ba2a231a67
Author: Kashif Faraz <[email protected]>
AuthorDate: Thu Oct 29 14:43:20 2026 +0530
fix: do not report poll-idle-ratio as 1.0 when it is unknown (#20783)
---
.../druid/indexing/kafka/KafkaConsumerMonitor.java | 6 ++-
.../SeekableStreamIndexTaskRunner.java | 23 +++++---
.../SeekableStreamIndexTaskRunnerTest.java | 61 ++++++++++++++++++++++
3 files changed, 81 insertions(+), 9 deletions(-)
diff --git
a/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaConsumerMonitor.java
b/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaConsumerMonitor.java
index 4a25c08d525..9015f8bee5a 100644
---
a/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaConsumerMonitor.java
+++
b/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaConsumerMonitor.java
@@ -214,6 +214,8 @@ public class KafkaConsumerMonitor extends AbstractMonitor
* sampling window (30s by default).
* A value of 0 represents that the consumer is never idle, i.e. always
consuming.
* A value of 1 represents that the consumer is always idle, i.e. not
receiving data.
+ * A negative value indicates that the metric is not yet known, e.g. because
the consumer
+ * has not reported it yet.
*/
public double getPollIdleRatioAvg()
{
@@ -223,6 +225,8 @@ public class KafkaConsumerMonitor extends AbstractMonitor
/**
* Cumulative average poll-to-idle ratio over the lifetime of the task
* as opposed to {@link #getPollIdleRatioAvg()} which only reflects Kafka's
own short sampling window.
+ * A negative value indicates that the metric is not yet known, e.g. because
the consumer
+ * has not reported any sample yet.
*/
public double getRunningAvgPollIdleRatio()
{
@@ -231,7 +235,7 @@ public class KafkaConsumerMonitor extends AbstractMonitor
private static class RunningAverage
{
- private static final double UNKNOWN = 1.0d;
+ private static final double UNKNOWN = -1.0d;
private double latest = UNKNOWN;
private double sum = 0.0d;
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java
index 4e92c47fe06..910940dd2f8 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java
@@ -1901,14 +1901,21 @@ public abstract class
SeekableStreamIndexTaskRunner<PartitionIdType, SequenceOff
returnMap.put("movingAverages", averagesMap);
returnMap.put("totals", totalsMap);
if (this.recordSupplier != null) {
- returnMap.put(
- AUTOSCALER_METRICS_KEY,
- Map.of(
- POLL_IDLE_RATIO_KEY, recordSupplier.getPollIdleRatioMetric(),
- POLL_IDLE_RATIO_RUNNING_AVG,
recordSupplier.getRunningAvgPollIdleRatio(),
- PROCESSING_RATE_RUNNING_AVG, getRunningAvgProcessingRate()
- )
- );
+ final Map<String, Object> autoscalerMetrics = new HashMap<>();
+ autoscalerMetrics.put(PROCESSING_RATE_RUNNING_AVG,
getRunningAvgProcessingRate());
+
+ // Only publish the poll-idle-ratio metrics once the recordSupplier has
a known value for them;
+ // otherwise an unknown value could pollute the aggregate average
computed across all tasks.
+ final double pollIdleRatio = recordSupplier.getPollIdleRatioMetric();
+ if (pollIdleRatio >= 0) {
+ autoscalerMetrics.put(POLL_IDLE_RATIO_KEY, pollIdleRatio);
+ }
+ final double runningAvgPollIdleRatio =
recordSupplier.getRunningAvgPollIdleRatio();
+ if (runningAvgPollIdleRatio >= 0) {
+ autoscalerMetrics.put(POLL_IDLE_RATIO_RUNNING_AVG,
runningAvgPollIdleRatio);
+ }
+
+ returnMap.put(AUTOSCALER_METRICS_KEY, autoscalerMetrics);
}
return returnMap;
}
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunnerTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunnerTest.java
index d6213864d52..6cca4ed94e6 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunnerTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunnerTest.java
@@ -279,6 +279,67 @@ public class SeekableStreamIndexTaskRunnerTest
Assertions.assertEquals(ImmutableMap.of("partition", "6"),
sequences.get(0).getEndOffsets());
}
+ @Test
+ public void testDoGetRowStatsOmitsPollIdleRatioUntilKnown()
+ {
+ final TestSeekableStreamIndexTaskRunner runner = createRunner();
+ Mockito.when(task.getId()).thenReturn("task1");
+ Mockito.when(task.getSupervisorId()).thenReturn("supervisorId");
+
+ // The recordSupplier reports no known poll-idle-ratio sample yet
(negative sentinel).
+ final RecordSupplier<?, ?, ?> recordSupplier =
Mockito.mock(RecordSupplier.class);
+ Mockito.when(recordSupplier.getPollIdleRatioMetric()).thenReturn(-1.0);
+ Mockito.when(recordSupplier.getRunningAvgPollIdleRatio()).thenReturn(-1.0);
+ Mockito.when(task.newTaskRecordSupplier(any()))
+ .thenReturn(recordSupplier);
+
+ final StreamAppenderator appenderator =
Mockito.mock(StreamAppenderator.class);
+ Mockito.when(task.newAppenderator(any(), any(), any(), any()))
+ .thenReturn(appenderator);
+
+ final List<DataSegment> segment = CreateDataSegments
+ .ofDatasource(DATA_SOURCE)
+ .withNumPartitions(1)
+ .withNumRows(10)
+ .eachOfSizeInMb(500);
+ final SegmentsAndCommitMetadata commitMetadata =
+ new SegmentsAndCommitMetadata(segment,
"offset-100").withWasPublished(true);
+
+ final StreamAppenderatorDriver driver =
Mockito.mock(StreamAppenderatorDriver.class);
+ Mockito.when(task.newDriver(any(), any(), any()))
+ .thenReturn(driver);
+ Mockito.when(driver.publish(any(), any(), any(), any()))
+ .thenReturn(Futures.immediateFuture(commitMetadata));
+ Mockito.when(driver.registerHandoff(any()))
+ .thenReturn(Futures.immediateFuture(commitMetadata));
+
+ runner.run(createTaskToolbox());
+
+ Map<String, Object> autoscalerMetrics =
+ (Map<String, Object>)
runner.doGetRowStats().get(SeekableStreamIndexTaskRunner.AUTOSCALER_METRICS_KEY);
+ Assertions.assertFalse(
+
autoscalerMetrics.containsKey(SeekableStreamIndexTaskRunner.POLL_IDLE_RATIO_KEY),
+ "pollIdleRatio must not be published until the recordSupplier reports
a known value"
+ );
+ Assertions.assertFalse(
+
autoscalerMetrics.containsKey(SeekableStreamIndexTaskRunner.POLL_IDLE_RATIO_RUNNING_AVG),
+ "pollIdleRatioRunningAvg must not be published until the
recordSupplier reports a known value"
+ );
+ Assertions.assertTrue(
+
autoscalerMetrics.containsKey(SeekableStreamIndexTaskRunner.PROCESSING_RATE_RUNNING_AVG),
+ "processingRateRunningAvg should always be published"
+ );
+
+ // Once the recordSupplier reports known values, they must be published.
+ Mockito.when(recordSupplier.getPollIdleRatioMetric()).thenReturn(0.25);
+ Mockito.when(recordSupplier.getRunningAvgPollIdleRatio()).thenReturn(0.4);
+
+ autoscalerMetrics =
+ (Map<String, Object>)
runner.doGetRowStats().get(SeekableStreamIndexTaskRunner.AUTOSCALER_METRICS_KEY);
+ Assertions.assertEquals(0.25,
autoscalerMetrics.get(SeekableStreamIndexTaskRunner.POLL_IDLE_RATIO_KEY));
+ Assertions.assertEquals(0.4,
autoscalerMetrics.get(SeekableStreamIndexTaskRunner.POLL_IDLE_RATIO_RUNNING_AVG));
+ }
+
@Test
public void testWithinMinMaxTime()
{
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
