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 c58e62b6be3 minor: Use running average values for processing rate and
poll idle ratio (#20708)
c58e62b6be3 is described below
commit c58e62b6be346ef505ba30f8b13e041725059a9b
Author: Kashif Faraz <[email protected]>
AuthorDate: Wed Oct 28 17:33:23 2026 +0530
minor: Use running average values for processing rate and poll idle ratio
(#20708)
Changes:
- Use running average metrics for Kafka poll-idle-ratio and average
processing rate (records / second)
- The average is taken over the entire lifecycle of a task and is not
exponentially weighted on purpose.
- This makes it easy to reason about the metric, and is driven by the
dominant behavior in the task duration.
- The poll-idle-ratio is reported by the Kafka consumer but is averaged
over a 30s window by default.
This often leads to volatile metric values and the value at the exact time
of the decision wins.
In production setups, this is a die roll since even though the metric had
been consistently reporting
a certain value, a momentary spike or dip would cause the auto-scaler to
make a bad decision.
- The `usePollIdleRatio: false` setting was also unusable before this patch
due to avg processing rate metric
being volatile.
---
.../CostBasedAutoScalerIntegrationTest.java | 2 +-
.../druid/indexing/kafka/KafkaConsumerMonitor.java | 47 +++++++++++++--
.../druid/indexing/kafka/KafkaRecordSupplier.java | 6 ++
.../SeekableStreamIndexTaskRunner.java | 68 +++++++++++++++++++++-
.../seekablestream/common/RecordSupplier.java | 9 +++
.../supervisor/autoscaler/CostBasedAutoScaler.java | 60 +++++++++++++++++--
.../autoscaler/CostBasedAutoScalerTest.java | 22 ++++---
7 files changed, 189 insertions(+), 25 deletions(-)
diff --git
a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/autoscaler/CostBasedAutoScalerIntegrationTest.java
b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/autoscaler/CostBasedAutoScalerIntegrationTest.java
index 879d90e6956..ae61b30859f 100644
---
a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/autoscaler/CostBasedAutoScalerIntegrationTest.java
+++
b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/autoscaler/CostBasedAutoScalerIntegrationTest.java
@@ -255,6 +255,7 @@ public class CostBasedAutoScalerIntegrationTest extends
StreamIndexTestBase
.taskCountMax(4)
.lagWeight(0.5)
.idleWeight(0.5)
+ .usePollIdleRatio(false)
.scaleActionPeriodMillis(10L)
.minScaleDownDelay(Duration.standardSeconds(1))
.build();
@@ -295,7 +296,6 @@ public class CostBasedAutoScalerIntegrationTest extends
StreamIndexTestBase
.hasDimension(DruidMetrics.SUPERVISOR_ID,
supervisor.getId())
.hasValueMatching(Matchers.equalTo(1L))
);
- Assertions.assertEquals(1, getCurrentTaskCount(supervisor.getId()));
cluster.callApi().postSupervisor(supervisor.createSuspendedSpec());
cluster.callApi().waitForAllSegmentsToBeAvailable(dataSource, coordinator,
broker);
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 08f933a31b0..4a25c08d525 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
@@ -19,7 +19,6 @@
package org.apache.druid.indexing.kafka;
-import com.google.common.util.concurrent.AtomicDouble;
import org.apache.druid.error.DruidException;
import org.apache.druid.java.util.emitter.service.ServiceEmitter;
import org.apache.druid.java.util.emitter.service.ServiceMetricEvent;
@@ -144,7 +143,7 @@ public class KafkaConsumerMonitor extends AbstractMonitor
private final Supplier<ServiceMetricEvent.Builder> metricBuilderSupplier;
private final Map<MetricName, AtomicLong> counters = new HashMap<>();
- private final AtomicDouble pollIdleRatioAvg = new AtomicDouble(1.0d);
+ private final RunningAverage pollIdleRatio = new RunningAverage();
public KafkaConsumerMonitor(
final KafkaConsumer<?, ?> consumer,
@@ -197,7 +196,7 @@ public class KafkaConsumerMonitor extends AbstractMonitor
// Capture `poll-idle-ratio-avg` metric for autoscaler purposes.
if (POLL_IDLE_RATIO_METRIC_NAME.equals(metricName.name())) {
if (entry.getValue().metricValue() != null) {
- pollIdleRatioAvg.set(((Number)
entry.getValue().metricValue()).doubleValue());
+ pollIdleRatio.addSample(((Number)
entry.getValue().metricValue()).doubleValue());
}
}
}
@@ -211,12 +210,50 @@ public class KafkaConsumerMonitor extends AbstractMonitor
}
/**
- * Average poll-to-idle ratio as reported by the Kafka consumer.
+ * poll-to-idle ratio as reported by the Kafka consumer averaged over a short
+ * 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.
*/
public double getPollIdleRatioAvg()
{
- return pollIdleRatioAvg.get();
+ return pollIdleRatio.getLatest();
+ }
+
+ /**
+ * 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.
+ */
+ public double getRunningAvgPollIdleRatio()
+ {
+ return pollIdleRatio.getAvg();
+ }
+
+ private static class RunningAverage
+ {
+ private static final double UNKNOWN = 1.0d;
+
+ private double latest = UNKNOWN;
+ private double sum = 0.0d;
+ private long count = 0L;
+
+ synchronized void addSample(final double value)
+ {
+ if (!Double.isNaN(value) && value >= 0 && value <= 1) {
+ sum += value;
+ count++;
+ }
+ latest = value;
+ }
+
+ synchronized double getAvg()
+ {
+ return count == 0 ? UNKNOWN : sum / count;
+ }
+
+ synchronized double getLatest()
+ {
+ return latest;
+ }
}
}
diff --git
a/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaRecordSupplier.java
b/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaRecordSupplier.java
index c20aa81f2c7..a1ff07d6ddb 100644
---
a/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaRecordSupplier.java
+++
b/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaRecordSupplier.java
@@ -306,6 +306,12 @@ public class KafkaRecordSupplier implements
RecordSupplier<KafkaTopicPartition,
return monitor.getPollIdleRatioAvg();
}
+ @Override
+ public double getRunningAvgPollIdleRatio()
+ {
+ return monitor.getRunningAvgPollIdleRatio();
+ }
+
@Override
public Set<KafkaTopicPartition> getPartitionIds(String stream)
{
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 1239f9f1c7e..4e92c47fe06 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
@@ -135,6 +135,7 @@ import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
@@ -152,6 +153,8 @@ public abstract class
SeekableStreamIndexTaskRunner<PartitionIdType, SequenceOff
{
public static final String AUTOSCALER_METRICS_KEY = "autoscalerMetrics";
public static final String POLL_IDLE_RATIO_KEY = "pollIdleRatio";
+ public static final String POLL_IDLE_RATIO_RUNNING_AVG =
"pollIdleRatioRunningAvg";
+ public static final String PROCESSING_RATE_RUNNING_AVG =
"processingRateRunningAvg";
private static final String CTX_KEY_LOOKUP_TIER = "lookupTier";
@@ -238,6 +241,18 @@ public abstract class
SeekableStreamIndexTaskRunner<PartitionIdType, SequenceOff
private volatile DateTime startTime;
private volatile Status status = Status.NOT_STARTED; // this is only ever
set by the task runner thread (runThread)
+
+ /**
+ * Tracks cumulative time spent in {@link Status#READING}, i.e. active
reading
+ * and processing records.
+ */
+ private final AtomicLong totalReadingMillis = new AtomicLong(0L);
+
+ /**
+ * A non-zero value implies that the task is currently in {@link
Status#READING}.
+ */
+ private final AtomicLong readingSinceMillis = new AtomicLong(0L);
+
private volatile TaskToolbox toolbox;
private volatile Thread runThread;
private volatile Appenderator appenderator;
@@ -635,6 +650,7 @@ public abstract class
SeekableStreamIndexTaskRunner<PartitionIdType, SequenceOff
// Could eventually support leader/follower mode (for keeping replicas
more in sync)
boolean stillReading = !assignment.isEmpty();
status = Status.READING;
+ markReadingStarted();
try {
while (stillReading) {
if (possiblyPause()) {
@@ -655,6 +671,7 @@ public abstract class
SeekableStreamIndexTaskRunner<PartitionIdType, SequenceOff
final SequenceMetadata<PartitionIdType, SequenceOffsetType>
latestSequence = getLastSequenceMetadataOrNull();
if (stopRequested.get() || latestSequence == null ||
latestSequence.isCheckpointed()) {
status = Status.PUBLISHING;
+ markReadingStopped();
}
if (stopRequested.get()) {
@@ -864,6 +881,7 @@ public abstract class
SeekableStreamIndexTaskRunner<PartitionIdType, SequenceOff
}
status = Status.PUBLISHING;
+ markReadingStopped();
}
// We need to copy sequences here, because the success callback in
publishAndRegisterHandoff removes items from
@@ -1527,6 +1545,7 @@ public abstract class
SeekableStreamIndexTaskRunner<PartitionIdType, SequenceOff
try {
if (pauseRequested) {
status = Status.PAUSED;
+ markReadingStopped();
hasPaused.signalAll();
long pauseTime = System.currentTimeMillis();
@@ -1536,6 +1555,7 @@ public abstract class
SeekableStreamIndexTaskRunner<PartitionIdType, SequenceOff
}
status = Status.READING;
+ markReadingStarted();
shouldResume.signalAll();
log.info("Received resume command, resuming ingestion.");
task.emitMetric(toolbox.getEmitter(), "ingest/pause/time",
System.currentTimeMillis() - pauseTime);
@@ -1554,6 +1574,28 @@ public abstract class
SeekableStreamIndexTaskRunner<PartitionIdType, SequenceOff
return status == Status.PAUSED;
}
+ /**
+ * Marks the start of a period of actively reading and processing records.
Must be paired with a subsequent
+ * call to {@link #markReadingStopped()} to accumulate {@link
#totalReadingMillis}.
+ */
+ private void markReadingStarted()
+ {
+ readingSinceMillis.compareAndSet(0L, System.currentTimeMillis());
+ }
+
+ /**
+ * Marks the end of a period of actively reading and processing records,
begun by the most recent call to
+ * {@link #markReadingStarted()}, and folds the elapsed duration into {@link
#totalReadingMillis}. A no-op if
+ * reading was not in progress.
+ */
+ private void markReadingStopped()
+ {
+ final long readingSinceMillis = this.readingSinceMillis.getAndSet(0L);
+ if (readingSinceMillis > 0) {
+ totalReadingMillis.addAndGet(System.currentTimeMillis() -
readingSinceMillis);
+ }
+ }
+
private void requestPause()
{
pauseRequested = true;
@@ -1859,11 +1901,35 @@ 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,
this.recordSupplier.getPollIdleRatioMetric()));
+ 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()
+ )
+ );
}
return returnMap;
}
+ /**
+ * Average number of records processed per second, over the cumulative time
spent actively reading and
+ * processing records (i.e. {@link Status#READING}), excluding startup,
paused and publishing time.
+ */
+ private double getRunningAvgProcessingRate()
+ {
+ long readingMillis = totalReadingMillis.get();
+ final long readingSince = readingSinceMillis.get();
+ if (readingSince > 0) {
+ readingMillis += System.currentTimeMillis() - readingSince;
+ }
+
+ return readingMillis <= 0
+ ? 0.0d
+ : rowIngestionMeters.getTotals().getProcessed() * 1000.0d /
readingMillis;
+ }
+
public Map<String, Object> doGetLiveReports()
{
Map<String, Object> ingestionStatsAndErrors = new HashMap<>();
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/common/RecordSupplier.java
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/common/RecordSupplier.java
index 7af4633102c..544ba9b2ea3 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/common/RecordSupplier.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/common/RecordSupplier.java
@@ -154,6 +154,15 @@ public interface RecordSupplier<PartitionIdType,
SequenceOffsetType, RecordType
return -1;
}
+ /**
+ * Running average of the {@link #getPollIdleRatioMetric()} averaged over the
+ * entire lifecycle of the task.
+ */
+ default double getRunningAvgPollIdleRatio()
+ {
+ return -1;
+ }
+
/**
* close the RecordSupplier
*/
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScaler.java
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScaler.java
index e34ad386099..60f8a643aa2 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScaler.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScaler.java
@@ -463,6 +463,30 @@ public class CostBasedAutoScaler implements
SupervisorTaskAutoScaler
return result.toIntArray();
}
+ /**
+ * Extracts the average processing rate (records per second) from the task
stats.
+ *
+ * @param taskStats the stats map from supervisor.getStats()
+ */
+ static double extractAvgProcessingRate(Map<String, Map<String, Object>>
taskStats)
+ {
+ if (taskStats == null || taskStats.isEmpty()) {
+ return -1.;
+ }
+
+ // First try using the running average if any of the tasks has started
reporting it
+ final double runningAverage = extractAutoscalerMetric(
+ taskStats,
+ SeekableStreamIndexTaskRunner.PROCESSING_RATE_RUNNING_AVG
+ );
+ if (runningAverage >= 0) {
+ return runningAverage;
+ } else {
+ // Fall back to the old moving average metric
+ return extractMovingAverage(taskStats);
+ }
+ }
+
/**
* Extracts the average poll-idle-ratio metric from task stats.
* This metric indicates how much time the consumer spends idle waiting for
data.
@@ -476,6 +500,30 @@ public class CostBasedAutoScaler implements
SupervisorTaskAutoScaler
return -1.;
}
+ // First try using running average metric if any of the task has started
reporting it
+ final double runningAverage = extractAutoscalerMetric(
+ taskStats,
+ SeekableStreamIndexTaskRunner.POLL_IDLE_RATIO_RUNNING_AVG
+ );
+ if (runningAverage >= 0) {
+ return runningAverage;
+ } else {
+ // Fall back to the instantaneous metric
+ return extractAutoscalerMetric(
+ taskStats,
+ SeekableStreamIndexTaskRunner.POLL_IDLE_RATIO_KEY
+ );
+ }
+ }
+
+ /**
+ * Gets the value of the given metric averaged across all tasks that have
reported the metric.
+ */
+ private static double extractAutoscalerMetric(
+ Map<String, Map<String, Object>> taskStats,
+ String metricName
+ )
+ {
double sum = 0;
int count = 0;
for (Map<String, Object> groupMetrics : taskStats.values()) {
@@ -483,12 +531,12 @@ public class CostBasedAutoScaler implements
SupervisorTaskAutoScaler
if (taskMetric instanceof Map) {
Object autoScalerMetricsMap = ((Map<?, ?>)
taskMetric).get(SeekableStreamIndexTaskRunner.AUTOSCALER_METRICS_KEY);
if (autoScalerMetricsMap instanceof Map) {
- Object pollIdleRatioAvg = ((Map<?, ?>)
autoScalerMetricsMap).get(SeekableStreamIndexTaskRunner.POLL_IDLE_RATIO_KEY);
- if (pollIdleRatioAvg instanceof Number) {
- double taskPollIdleRatio = ((Number)
pollIdleRatioAvg).doubleValue();
+ Object metricValue = ((Map<?, ?>)
autoScalerMetricsMap).get(metricName);
+ if (metricValue instanceof Number) {
+ double metricDoubleValue = ((Number) metricValue).doubleValue();
// Do not include negative values
- if (taskPollIdleRatio >= 0) {
- sum += taskPollIdleRatio;
+ if (metricDoubleValue >= 0) {
+ sum += metricDoubleValue;
count++;
}
}
@@ -576,7 +624,7 @@ public class CostBasedAutoScaler implements
SupervisorTaskAutoScaler
final int partitionCount = supervisor.getPartitionCount();
final Map<String, Map<String, Object>> taskStats = supervisor.getStats();
- final double movingAvgRate = extractMovingAverage(taskStats);
+ final double movingAvgRate = extractAvgProcessingRate(taskStats);
final double pollIdleRatio = extractPollIdleRatio(taskStats);
if (!config.isUsePollIdleRatio() && aggregateLag > 0 && movingAvgRate > 0)
{
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScalerTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScalerTest.java
index 374ecd3bfc5..471ba73dfad 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScalerTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScalerTest.java
@@ -441,7 +441,7 @@ public class CostBasedAutoScalerTest
Map<String, Map<String, Object>> nonNumberRatio = new HashMap<>();
Map<String, Object> taskStats3 = new HashMap<>();
Map<String, Object> autoscalerMetrics = new HashMap<>();
- autoscalerMetrics.put(SeekableStreamIndexTaskRunner.POLL_IDLE_RATIO_KEY,
"not-a-number");
+
autoscalerMetrics.put(SeekableStreamIndexTaskRunner.POLL_IDLE_RATIO_RUNNING_AVG,
"not-a-number");
taskStats3.put(SeekableStreamIndexTaskRunner.AUTOSCALER_METRICS_KEY,
autoscalerMetrics);
nonNumberRatio.put("0", Collections.singletonMap("task-0", taskStats3));
Assertions.assertEquals(
@@ -487,7 +487,7 @@ public class CostBasedAutoScalerTest
validStats.put("0", group);
Assertions.assertEquals(
1500.0,
- CostBasedAutoScaler.extractMovingAverage(validStats),
+ CostBasedAutoScaler.extractAvgProcessingRate(validStats),
0.0001,
"Average 5-minute processing rate should be computed across tasks"
);
@@ -856,7 +856,7 @@ public class CostBasedAutoScalerTest
private Map<String, Object> buildTaskStatsWithPollIdle(double pollIdleRatio)
{
Map<String, Object> autoscalerMetrics = new HashMap<>();
- autoscalerMetrics.put(SeekableStreamIndexTaskRunner.POLL_IDLE_RATIO_KEY,
pollIdleRatio);
+
autoscalerMetrics.put(SeekableStreamIndexTaskRunner.POLL_IDLE_RATIO_RUNNING_AVG,
pollIdleRatio);
Map<String, Object> taskStats = new HashMap<>();
taskStats.put(SeekableStreamIndexTaskRunner.AUTOSCALER_METRICS_KEY,
autoscalerMetrics);
@@ -865,15 +865,13 @@ public class CostBasedAutoScalerTest
private Map<String, Object> buildTaskStatsWithMovingAverage(double
processedRate)
{
- Map<String, Object> buildSegments = new HashMap<>();
- buildSegments.put(FIVE_MINUTE_NAME, Map.of(RowIngestionMeters.PROCESSED,
processedRate));
-
- Map<String, Object> movingAverages = new HashMap<>();
- movingAverages.put(RowIngestionMeters.BUILD_SEGMENTS, buildSegments);
-
- Map<String, Object> taskStats = new HashMap<>();
- taskStats.put("movingAverages", movingAverages);
- return taskStats;
+ return Map.of(
+ SeekableStreamIndexTaskRunner.AUTOSCALER_METRICS_KEY,
+ Map.of(
+ SeekableStreamIndexTaskRunner.PROCESSING_RATE_RUNNING_AVG,
+ processedRate
+ )
+ );
}
private Map<String, Object>
buildTaskStatsWithMovingAverageForInterval(String intervalName, double
processedRate)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
