This is an automated email from the ASF dual-hosted git repository.
Fly-Style 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 6bed96efcb9 feat: introduce processing rate to max ratio as
poll-idle-ratio replacement (#19622)
6bed96efcb9 is described below
commit 6bed96efcb91afc1f0e9e7e324c7409a955bd1cd
Author: Sasha Syrotenko <[email protected]>
AuthorDate: Fri Jun 26 15:12:56 2026 +0300
feat: introduce processing rate to max ratio as poll-idle-ratio replacement
(#19622)
---
.../CostBasedAutoScalerIntegrationTest.java | 65 ++++++++-
.../supervisor/autoscaler/CostBasedAutoScaler.java | 18 ++-
.../autoscaler/CostBasedAutoScalerConfig.java | 30 ++++-
.../supervisor/autoscaler/CostMetrics.java | 35 ++++-
.../autoscaler/WeightedCostFunction.java | 21 +--
.../autoscaler/CostBasedAutoScalerConfigTest.java | 7 +-
.../autoscaler/CostBasedAutoScalerMockTest.java | 3 +-
.../autoscaler/CostBasedAutoScalerTest.java | 145 ++++++++++++++++-----
.../autoscaler/WeightedCostFunctionTest.java | 93 ++++++++++++-
9 files changed, 364 insertions(+), 53 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 7e63c3f4f25..90f78bec1ef 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
@@ -38,6 +38,7 @@ import org.hamcrest.Matchers;
import org.jboss.netty.handler.codec.http.HttpMethod;
import org.joda.time.Duration;
import org.joda.time.Period;
+import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -46,6 +47,7 @@ import org.junit.jupiter.api.Timeout;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import static
org.apache.druid.indexing.seekablestream.supervisor.autoscaler.CostBasedAutoScaler.OPTIMAL_TASK_COUNT_METRIC;
@@ -62,6 +64,7 @@ public class CostBasedAutoScalerIntegrationTest extends
StreamIndexTestBase
private String topic;
private final KafkaResource kafkaServer = new KafkaResource();
+ private ExecutorService backgroundPublishExecutor;
@Override
protected StreamIngestResource<?> getStreamIngestResource()
@@ -84,6 +87,14 @@ public class CostBasedAutoScalerIntegrationTest extends
StreamIndexTestBase
kafkaServer.createTopicWithPartitions(topic, PARTITION_COUNT);
}
+ @AfterEach
+ public void shutdownBackgroundPublishExecutor()
+ {
+ if (backgroundPublishExecutor != null) {
+ backgroundPublishExecutor.shutdownNow();
+ }
+ }
+
@Test
@Timeout(45)
public void test_autoScaler_computesOptimalTaskCountAndProduceScaleDown()
@@ -137,7 +148,8 @@ public class CostBasedAutoScalerIntegrationTest extends
StreamIndexTestBase
// Produce additional records to create a backlog / lag
// This ensures tasks are busy processing (low idle ratio)
- Executors.newSingleThreadExecutor().submit(() -> {
+ backgroundPublishExecutor = Executors.newSingleThreadExecutor();
+ backgroundPublishExecutor.submit(() -> {
for (int i = 0; i < 500; ++i) {
publish1kRecords(topic, true);
}
@@ -180,6 +192,55 @@ public class CostBasedAutoScalerIntegrationTest extends
StreamIndexTestBase
cluster.callApi().postSupervisor(kafkaSupervisorSpec.createSuspendedSpec());
}
+ @Test
+ public void
test_autoScaler_computesOptimalTaskCountAndProducesScaleUp_withUtilizationRatio()
+ {
+
+ final int lowInitialTaskCount = 1;
+ // This ensures tasks are busy processing (low idle ratio)
+ backgroundPublishExecutor = Executors.newSingleThreadExecutor();
+ backgroundPublishExecutor.submit(() -> {
+ for (int i = 0; i < 500; ++i) {
+ publish1kRecords(topic, true);
+ }
+ });
+
+ final CostBasedAutoScalerConfig autoScalerConfig =
CostBasedAutoScalerConfig
+ .builder()
+ .enableTaskAutoScaler(true)
+ .taskCountMin(1)
+ .taskCountMax(50)
+ .taskCountStart(lowInitialTaskCount)
+ .scaleActionPeriodMillis(500)
+ .minTriggerScaleActionFrequencyMillis(1000)
+ .lagWeight(0.8)
+ .idleWeight(0.2)
+ .usePollIdleRatio(false)
+ .build();
+
+ final KafkaSupervisorSpec kafkaSupervisorSpec =
createKafkaSupervisorWithAutoScaler(
+ autoScalerConfig,
+ lowInitialTaskCount
+ );
+
+ Assertions.assertEquals(kafkaSupervisorSpec.getId(),
cluster.callApi().postSupervisor(kafkaSupervisorSpec));
+
+ // Wait for the supervisor is running
+ overlord.latchableEmitter()
+ .waitForEvent(event -> event.hasMetricName("task/run/time")
+ .hasDimension(DruidMetrics.DATASOURCE,
dataSource));
+
+ // With 50 partitions and high lag saturating the single task's processing
rate,
+ // the utilization ratio must drive the cost function to recommend scaling
up.
+ overlord.latchableEmitter().waitForEvent(
+ event -> event.hasMetricName(OPTIMAL_TASK_COUNT_METRIC)
+ .hasValueMatching(Matchers.greaterThan(1L))
+ );
+
+ // Suspend the supervisor
+
cluster.callApi().postSupervisor(kafkaSupervisorSpec.createSuspendedSpec());
+ }
+
@Test
public void test_autoScaler_scalesUpAndDown_withSlowPublish()
{
@@ -337,7 +398,7 @@ public class CostBasedAutoScalerIntegrationTest extends
StreamIndexTestBase
final String getSupervisorPath =
StringUtils.format("/druid/indexer/v1/supervisor/%s", supervisorId);
final KafkaSupervisorSpec supervisorSpec =
cluster.callApi().serviceClient().onLeaderOverlord(
mapper -> new RequestBuilder(HttpMethod.GET, getSupervisorPath),
- new TypeReference<>(){}
+ new TypeReference<>() {}
);
Assertions.assertNotNull(supervisorSpec);
return supervisorSpec.getSpec().getIOConfig().getTaskCount();
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 32efeadd4c7..ed3454c929a 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
@@ -19,6 +19,7 @@
package org.apache.druid.indexing.seekablestream.supervisor.autoscaler;
+import com.google.common.collect.EvictingQueue;
import it.unimi.dsi.fastutil.ints.IntArraySet;
import it.unimi.dsi.fastutil.ints.IntSet;
import org.apache.druid.error.DruidException;
@@ -40,6 +41,7 @@ import org.apache.druid.utils.CollectionUtils;
import javax.annotation.Nullable;
import java.util.Arrays;
+import java.util.Collections;
import java.util.Map;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
@@ -80,6 +82,12 @@ public class CostBasedAutoScaler implements
SupervisorTaskAutoScaler
*/
static final int MAX_IDLENESS_PARTITION_LAG = 10_000;
+ /**
+ * The number of most recent processing rate samples to maintain for
auto-scaling decisions
+ * in case when {@link CostBasedAutoScalerConfig#usePollIdleRatio} is
disabled.
+ */
+ static final int PROCESSING_RATE_WINDOW_SIZE = 50;
+
private final String supervisorId;
private final SeekableStreamSupervisor supervisor;
private final ServiceEmitter emitter;
@@ -87,6 +95,8 @@ public class CostBasedAutoScaler implements
SupervisorTaskAutoScaler
private final CostBasedAutoScalerConfig config;
private final ScheduledExecutorService autoscalerExecutor;
private final WeightedCostFunction costFunction;
+
+ private final EvictingQueue<Double> processingRateSamples;
private volatile CostMetrics lastKnownMetrics;
public CostBasedAutoScaler(
@@ -102,6 +112,7 @@ public class CostBasedAutoScaler implements
SupervisorTaskAutoScaler
this.supervisorId = spec.getId();
this.emitter = emitter;
+ this.processingRateSamples =
EvictingQueue.create(PROCESSING_RATE_WINDOW_SIZE);
this.costFunction = new WeightedCostFunction();
this.autoscalerExecutor =
Execs.scheduledSingleThreaded("CostBasedAutoScaler-"
+
StringUtils.encodeForFormat(spec.getId()));
@@ -438,13 +449,18 @@ public class CostBasedAutoScaler implements
SupervisorTaskAutoScaler
final double movingAvgRate = extractMovingAverage(taskStats);
final double pollIdleRatio = extractPollIdleRatio(taskStats);
+ if (!config.isUsePollIdleRatio() && movingAvgRate > 0) {
+ processingRateSamples.add(movingAvgRate);
+ }
+
return new CostMetrics(
avgPartitionLag,
currentTaskCount,
partitionCount,
pollIdleRatio,
supervisor.getIoConfig().getTaskDuration().getStandardSeconds(),
- movingAvgRate
+ movingAvgRate,
+ processingRateSamples.isEmpty() ? null :
Collections.max(processingRateSamples)
);
}
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScalerConfig.java
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScalerConfig.java
index 7987c935b8e..167ffcaf9c7 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScalerConfig.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScalerConfig.java
@@ -67,6 +67,7 @@ public class CostBasedAutoScalerConfig implements
AutoScalerConfig
private final Duration minScaleUpDelay;
private final Duration minScaleDownDelay;
private final boolean scaleDownDuringTaskRolloverOnly;
+ private final boolean usePollIdleRatio;
/**
* Creates a new CostBasedAutoScalerConfig instance.
@@ -89,7 +90,8 @@ public class CostBasedAutoScalerConfig implements
AutoScalerConfig
@Nullable @JsonProperty("useTaskCountBoundariesOnScaleDown") Boolean
useTaskCountBoundariesOnScaleDown,
@Nullable @JsonProperty("minScaleUpDelay") Duration minScaleUpDelay,
@Nullable @JsonProperty("minScaleDownDelay") Duration minScaleDownDelay,
- @Nullable @JsonProperty("scaleDownDuringTaskRolloverOnly") Boolean
scaleDownDuringTaskRolloverOnly
+ @Nullable @JsonProperty("scaleDownDuringTaskRolloverOnly") Boolean
scaleDownDuringTaskRolloverOnly,
+ @Nullable @JsonProperty("usePollIdleRatio") Boolean usePollIdleRatio
)
{
this.enableTaskAutoScaler = enableTaskAutoScaler != null ?
enableTaskAutoScaler : false;
@@ -114,6 +116,7 @@ public class CostBasedAutoScalerConfig implements
AutoScalerConfig
);
this.minScaleDownDelay = Configs.valueOrDefault(minScaleDownDelay,
DEFAULT_MIN_SCALE_DELAY);
this.scaleDownDuringTaskRolloverOnly =
Configs.valueOrDefault(scaleDownDuringTaskRolloverOnly, false);
+ this.usePollIdleRatio = Configs.valueOrDefault(usePollIdleRatio, true);
if (this.enableTaskAutoScaler) {
Preconditions.checkNotNull(taskCountMax, "taskCountMax is required when
enableTaskAutoScaler is true");
@@ -272,6 +275,16 @@ public class CostBasedAutoScalerConfig implements
AutoScalerConfig
return scaleDownDuringTaskRolloverOnly;
}
+ /**
+ * If true (the default), idle cost is derived from the poll idle ratio
reported by tasks.
+ * If false, idle cost is derived from processing-rate utilization instead.
+ */
+ @JsonProperty
+ public boolean isUsePollIdleRatio()
+ {
+ return usePollIdleRatio;
+ }
+
@Override
public SupervisorTaskAutoScaler createAutoScaler(Supervisor supervisor,
SupervisorSpec spec, ServiceEmitter emitter)
{
@@ -302,6 +315,7 @@ public class CostBasedAutoScalerConfig implements
AutoScalerConfig
&& Objects.equals(minScaleUpDelay, that.minScaleUpDelay)
&& Objects.equals(minScaleDownDelay, that.minScaleDownDelay)
&& scaleDownDuringTaskRolloverOnly ==
that.scaleDownDuringTaskRolloverOnly
+ && usePollIdleRatio == that.usePollIdleRatio
&& Objects.equals(taskCountStart, that.taskCountStart)
&& Objects.equals(stopTaskCountRatio, that.stopTaskCountRatio);
}
@@ -323,7 +337,8 @@ public class CostBasedAutoScalerConfig implements
AutoScalerConfig
useTaskCountBoundariesOnScaleDown,
minScaleUpDelay,
minScaleDownDelay,
- scaleDownDuringTaskRolloverOnly
+ scaleDownDuringTaskRolloverOnly,
+ usePollIdleRatio
);
}
@@ -345,6 +360,7 @@ public class CostBasedAutoScalerConfig implements
AutoScalerConfig
", minScaleUpDelay=" + minScaleUpDelay +
", minScaleDownDelay=" + minScaleDownDelay +
", scaleDownDuringTaskRolloverOnly=" +
scaleDownDuringTaskRolloverOnly +
+ ", usePollIdleRatio=" + usePollIdleRatio +
'}';
}
@@ -368,6 +384,7 @@ public class CostBasedAutoScalerConfig implements
AutoScalerConfig
private Duration minScaleUpDelay;
private Duration minScaleDownDelay;
private Boolean scaleDownDuringTaskRolloverOnly;
+ private Boolean usePollIdleRatio;
private Builder()
{
@@ -445,6 +462,12 @@ public class CostBasedAutoScalerConfig implements
AutoScalerConfig
return this;
}
+ public Builder usePollIdleRatio(boolean usePollIdleRatio)
+ {
+ this.usePollIdleRatio = usePollIdleRatio;
+ return this;
+ }
+
public Builder useTaskCountBoundariesOnScaleUp(boolean
useTaskCountBoundariesOnScaleUp)
{
this.useTaskCountBoundariesOnScaleUp = useTaskCountBoundariesOnScaleUp;
@@ -473,7 +496,8 @@ public class CostBasedAutoScalerConfig implements
AutoScalerConfig
useTaskCountBoundariesOnScaleDown,
minScaleUpDelay,
minScaleDownDelay,
- scaleDownDuringTaskRolloverOnly
+ scaleDownDuringTaskRolloverOnly,
+ usePollIdleRatio
);
}
}
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostMetrics.java
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostMetrics.java
index 8938ad11b96..33bf7eff33d 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostMetrics.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostMetrics.java
@@ -19,6 +19,7 @@
package org.apache.druid.indexing.seekablestream.supervisor.autoscaler;
+import javax.annotation.Nullable;
import java.util.Objects;
/**
@@ -35,6 +36,8 @@ public class CostMetrics
private final long taskDurationSeconds;
private final double avgProcessingRate;
private final double aggregateLag;
+ @Nullable
+ private final Double maxObservedRate;
public CostMetrics(
double avgPartitionLag,
@@ -42,7 +45,8 @@ public class CostMetrics
int partitionCount,
double pollIdleRatio,
long taskDurationSeconds,
- double avgProcessingRate
+ double avgProcessingRate,
+ @Nullable Double maxObservedRate
)
{
this.avgPartitionLag = avgPartitionLag;
@@ -52,6 +56,7 @@ public class CostMetrics
this.taskDurationSeconds = taskDurationSeconds;
this.avgProcessingRate = avgProcessingRate;
this.aggregateLag = avgPartitionLag * partitionCount;
+ this.maxObservedRate = maxObservedRate;
}
/**
@@ -105,6 +110,27 @@ public class CostMetrics
return avgProcessingRate;
}
+ @Nullable
+ public Double getMaxObservedRate()
+ {
+ return maxObservedRate;
+ }
+
+
+ /**
+ * Derives the current idle ratio from the processing rate relative to the
max observed rate
+ * ({@code avgProcessingRate / maxObservedRate}). Returns a negative value
when no rate baseline
+ * is available yet (no observed rate samples, or a zero watermark),
signalling an unknown idle ratio.
+ */
+ double estimateIdleRatioFromProcessingRate()
+ {
+ if (maxObservedRate == null || maxObservedRate <= 0) {
+ return -1.0;
+ }
+ final double ratio = Math.min(1.0, avgProcessingRate / maxObservedRate);
+ return 1.0 - ratio;
+ }
+
@Override
public boolean equals(Object o)
{
@@ -120,7 +146,8 @@ public class CostMetrics
&& partitionCount == that.partitionCount
&& Double.compare(that.pollIdleRatio, pollIdleRatio) == 0
&& taskDurationSeconds == that.taskDurationSeconds
- && Double.compare(that.avgProcessingRate, avgProcessingRate) == 0;
+ && Double.compare(that.avgProcessingRate, avgProcessingRate) == 0
+ && Objects.equals(that.maxObservedRate, maxObservedRate);
}
@Override
@@ -132,7 +159,8 @@ public class CostMetrics
partitionCount,
pollIdleRatio,
taskDurationSeconds,
- avgProcessingRate
+ avgProcessingRate,
+ maxObservedRate
);
}
@@ -146,6 +174,7 @@ public class CostMetrics
", pollIdleRatio=" + pollIdleRatio +
", taskDurationSeconds=" + taskDurationSeconds +
", avgProcessingRate=" + avgProcessingRate +
+ ", maxObservedRate=" + maxObservedRate +
'}';
}
}
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/WeightedCostFunction.java
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/WeightedCostFunction.java
index ee63dbe870e..3e9d7e81b8f 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/WeightedCostFunction.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/WeightedCostFunction.java
@@ -117,26 +117,27 @@ public class WeightedCostFunction
lagRecoveryTime = metrics.getAggregateLag() * amplification /
(proposedTaskCount * adjustedProcessingRate);
}
- // Capacity-based idle prediction with sublinear busy redistribution. When
the proposed count
- // would oversaturate the cluster (busy work exceeds available capacity),
the unmet demand
- // becomes a virtual lag-recovery time on the same axis as real lag — so
the optimizer treats
- // predicted saturation as predicted lag, not as "perfect utilization".
- final double currentPollIdleRatio = metrics.getPollIdleRatio();
+ // Capacity-based idle prediction. When the proposed count would
oversaturate the cluster
+ // (busy work exceeds available capacity), the unmet demand becomes a
virtual lag-recovery
+ // time on the same axis as real lag — so the optimizer treats predicted
saturation as predicted lag.
+ final double currentIdleRatio = config.isUsePollIdleRatio()
+ ? metrics.getPollIdleRatio()
+ :
metrics.estimateIdleRatioFromProcessingRate();
final int currentTaskCount = metrics.getCurrentTaskCount();
final double predictedIdleRatio;
final double overrun;
- if (currentPollIdleRatio < 0) {
+ if (currentIdleRatio < 0) {
predictedIdleRatio = 0.5;
overrun = 0.0;
} else if (currentTaskCount <= 0 || proposedTaskCount == currentTaskCount)
{
- predictedIdleRatio = currentPollIdleRatio;
+ predictedIdleRatio = currentIdleRatio;
overrun = 0.0;
} else {
- final double busyFraction = 1.0 - currentPollIdleRatio;
+ final double busyFraction = 1.0 - currentIdleRatio;
// Sublinear redistribution factor: keeps a healthy consolidation from
projecting negative idle,
// yet still diverges as proposed -> 0 so extreme consolidation
registers as overrun.
- final double idle = Math.pow((double) currentTaskCount /
proposedTaskCount, IDLE_SUBLINEARITY_EXPONENT);
- final double rawIdle = 1.0 - busyFraction * idle;
+ final double taskRatio = Math.pow((double) currentTaskCount /
proposedTaskCount, IDLE_SUBLINEARITY_EXPONENT);
+ final double rawIdle = 1.0 - busyFraction * taskRatio;
if (rawIdle >= 0) {
predictedIdleRatio = Math.min(1.0, rawIdle);
overrun = 0.0;
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScalerConfigTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScalerConfigTest.java
index 5a1c6ec98fc..6a8eec3e3d9 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScalerConfigTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScalerConfigTest.java
@@ -51,7 +51,8 @@ public class CostBasedAutoScalerConfigTest
+ " \"idleWeight\": 0.4,\n"
+ " \"minScaleUpDelay\": \"PT5M\",\n"
+ " \"minScaleDownDelay\": \"PT10M\",\n"
- + " \"scaleDownDuringTaskRolloverOnly\": true\n"
+ + " \"scaleDownDuringTaskRolloverOnly\": true,\n"
+ + " \"usePollIdleRatio\": false\n"
+ "}";
CostBasedAutoScalerConfig config = mapper.readValue(json,
CostBasedAutoScalerConfig.class);
@@ -67,6 +68,7 @@ public class CostBasedAutoScalerConfigTest
Assert.assertEquals(Duration.standardMinutes(5),
config.getMinScaleUpDelay());
Assert.assertEquals(Duration.standardMinutes(10),
config.getMinScaleDownDelay());
Assert.assertTrue(config.isScaleDownOnTaskRolloverOnly());
+ Assert.assertFalse(config.isUsePollIdleRatio());
Assert.assertFalse(config.isUseTaskCountBoundariesOnScaleUp());
Assert.assertTrue(config.isUseTaskCountBoundariesOnScaleDown());
@@ -101,6 +103,7 @@ public class CostBasedAutoScalerConfigTest
Assert.assertEquals(Duration.millis(DEFAULT_SCALE_ACTION_PERIOD_MILLIS),
config.getMinScaleUpDelay());
Assert.assertEquals(DEFAULT_MIN_SCALE_DELAY,
config.getMinScaleDownDelay());
Assert.assertFalse(config.isScaleDownOnTaskRolloverOnly());
+ Assert.assertTrue(config.isUsePollIdleRatio());
Assert.assertFalse(config.isUseTaskCountBoundariesOnScaleUp());
Assert.assertTrue(config.isUseTaskCountBoundariesOnScaleDown());
Assert.assertNull(config.getTaskCountStart());
@@ -190,6 +193,7 @@ public class CostBasedAutoScalerConfigTest
.minScaleUpDelay(Duration.standardMinutes(5))
.minScaleDownDelay(Duration.standardMinutes(10))
.scaleDownDuringTaskRolloverOnly(true)
+
.usePollIdleRatio(false)
.build();
Assert.assertTrue(config.getEnableTaskAutoScaler());
@@ -205,6 +209,7 @@ public class CostBasedAutoScalerConfigTest
Assert.assertEquals(Duration.standardMinutes(5),
config.getMinScaleUpDelay());
Assert.assertEquals(Duration.standardMinutes(10),
config.getMinScaleDownDelay());
Assert.assertTrue(config.isScaleDownOnTaskRolloverOnly());
+ Assert.assertFalse(config.isUsePollIdleRatio());
}
@Test
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScalerMockTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScalerMockTest.java
index 129569864f9..fa86fbf1d3c 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScalerMockTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScalerMockTest.java
@@ -371,7 +371,8 @@ public class CostBasedAutoScalerMockTest
PARTITION_COUNT,
pollIdleRatio,
TASK_DURATION_SECONDS,
- AVG_PROCESSING_RATE
+ AVG_PROCESSING_RATE,
+ 0.
);
doReturn(metrics).when(autoScaler).collectMetrics();
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 e21375a1b93..d773e20bfcb 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
@@ -62,6 +62,7 @@ public class CostBasedAutoScalerTest
when(mockSupervisorSpec.getDataSources()).thenReturn(List.of("test-datasource"));
when(mockSupervisor.getIoConfig()).thenReturn(mockIoConfig);
when(mockIoConfig.getStream()).thenReturn("test-stream");
+ when(mockIoConfig.getTaskDuration()).thenReturn(Duration.standardHours(1));
CostBasedAutoScalerConfig config = CostBasedAutoScalerConfig.builder()
.taskCountMax(100)
@@ -190,14 +191,16 @@ public class CostBasedAutoScalerTest
@Test
public void testComputeOptimalTaskCountLimitsTaskCountJumps()
{
- final CostBasedAutoScalerConfig boundedScaleUpConfig =
CostBasedAutoScalerConfig.builder()
-
.taskCountMax(100)
-
.taskCountMin(1)
-
.enableTaskAutoScaler(true)
-
.lagWeight(1.0)
-
.idleWeight(0.0)
-
.useTaskCountBoundariesOnScaleUp(true)
-
.build();
+ final CostBasedAutoScalerConfig boundedScaleUpConfig =
CostBasedAutoScalerConfig
+ .builder()
+ .taskCountMax(100)
+ .taskCountMin(1)
+ .enableTaskAutoScaler(true)
+ .lagWeight(1.0)
+ .idleWeight(0.0)
+ .useTaskCountBoundariesOnScaleUp(
+ true)
+ .build();
final CostBasedAutoScaler boundedScaleUpScaler =
createAutoScaler(boundedScaleUpConfig);
Assert.assertEquals(
@@ -206,13 +209,14 @@ public class CostBasedAutoScalerTest
boundedScaleUpScaler.computeOptimalTaskCount(createMetrics(100_000.0,
10, 100, 0.25))
);
- final CostBasedAutoScalerConfig unboundedScaleUpConfig =
CostBasedAutoScalerConfig.builder()
-
.taskCountMax(100)
-
.taskCountMin(1)
-
.enableTaskAutoScaler(true)
-
.lagWeight(1.0)
-
.idleWeight(0.0)
-
.build();
+ final CostBasedAutoScalerConfig unboundedScaleUpConfig =
CostBasedAutoScalerConfig
+ .builder()
+ .taskCountMax(100)
+ .taskCountMin(1)
+ .enableTaskAutoScaler(true)
+ .lagWeight(1.0)
+ .idleWeight(0.0)
+ .build();
final CostBasedAutoScaler unboundedScaleUpScaler =
createAutoScaler(unboundedScaleUpConfig);
Assert.assertEquals(
"Without scale-up boundaries, lag-only optimization should jump to max
task count",
@@ -220,14 +224,16 @@ public class CostBasedAutoScalerTest
unboundedScaleUpScaler.computeOptimalTaskCount(createMetrics(100_000.0, 10,
100, 0.25))
);
- final CostBasedAutoScalerConfig boundedScaleDownConfig =
CostBasedAutoScalerConfig.builder()
-
.taskCountMax(100)
-
.taskCountMin(1)
-
.enableTaskAutoScaler(true)
-
.lagWeight(0.0)
-
.idleWeight(1.0)
-
.useTaskCountBoundariesOnScaleDown(true)
-
.build();
+ final CostBasedAutoScalerConfig boundedScaleDownConfig =
CostBasedAutoScalerConfig
+ .builder()
+ .taskCountMax(100)
+ .taskCountMin(1)
+ .enableTaskAutoScaler(true)
+ .lagWeight(0.0)
+ .idleWeight(1.0)
+ .useTaskCountBoundariesOnScaleDown(
+ true)
+ .build();
final CostBasedAutoScaler boundedScaleDownScaler =
createAutoScaler(boundedScaleDownConfig);
Assert.assertEquals(
@@ -236,13 +242,14 @@ public class CostBasedAutoScalerTest
boundedScaleDownScaler.computeOptimalTaskCount(createMetrics(0.0, 100,
100, 0.9))
);
- final CostBasedAutoScalerConfig unboundedScaleDownConfig =
CostBasedAutoScalerConfig.builder()
-
.taskCountMax(25)
-
.taskCountMin(1)
-
.enableTaskAutoScaler(true)
-
.lagWeight(0.0)
-
.idleWeight(1.0)
-
.build();
+ final CostBasedAutoScalerConfig unboundedScaleDownConfig =
CostBasedAutoScalerConfig
+ .builder()
+ .taskCountMax(25)
+ .taskCountMin(1)
+ .enableTaskAutoScaler(true)
+ .lagWeight(0.0)
+ .idleWeight(1.0)
+ .build();
final CostBasedAutoScaler unboundedScaleDownScaler =
createAutoScaler(unboundedScaleDownConfig);
Assert.assertEquals(
"Without scale-down boundaries, idle-only optimization may select a
much lower task count",
@@ -497,6 +504,7 @@ public class CostBasedAutoScalerTest
when(spec.getDataSources()).thenReturn(List.of("test-datasource"));
when(supervisor.getIoConfig()).thenReturn(ioConfig);
when(ioConfig.getStream()).thenReturn("stream");
+ when(ioConfig.getTaskDuration()).thenReturn(Duration.standardHours(1));
CostBasedAutoScalerConfig cfgWithDefaults =
CostBasedAutoScalerConfig.builder()
.taskCountMax(10)
@@ -563,6 +571,79 @@ public class CostBasedAutoScalerTest
);
}
+ @Test
+ public void
testCollectMetricsTracksMaxProcessingRateOnlyWhenPollIdleRatioDisabled()
+ {
+ SupervisorSpec spec = Mockito.mock(SupervisorSpec.class);
+ SeekableStreamSupervisor supervisor =
Mockito.mock(SeekableStreamSupervisor.class);
+ ServiceEmitter emitter = Mockito.mock(ServiceEmitter.class);
+ SeekableStreamSupervisorIOConfig ioConfig =
Mockito.mock(SeekableStreamSupervisorIOConfig.class);
+
+ when(spec.getId()).thenReturn("test-supervisor");
+ when(spec.getDataSources()).thenReturn(List.of("test-datasource"));
+ when(spec.isSuspended()).thenReturn(false);
+ when(supervisor.getIoConfig()).thenReturn(ioConfig);
+ when(ioConfig.getStream()).thenReturn("test-stream");
+ when(ioConfig.getTaskDuration()).thenReturn(Duration.standardHours(1));
+ when(supervisor.getPartitionCount()).thenReturn(1);
+ when(supervisor.computeLagStats()).thenReturn(new LagStats(0, 0, 0));
+
+ // usePollIdleRatio defaults to true, which disables rate-watermark
tracking entirely.
+ CostBasedAutoScalerConfig defaultConfig =
CostBasedAutoScalerConfig.builder()
+
.taskCountMax(10)
+
.taskCountMin(1)
+
.enableTaskAutoScaler(true)
+
.build();
+ when(supervisor.getStats()).thenReturn(buildTaskStatsForRate(500.0));
+ CostBasedAutoScaler defaultScaler = new CostBasedAutoScaler(supervisor,
defaultConfig, spec, emitter);
+ Assert.assertNull(
+ "With usePollIdleRatio=true (the default), samples are never
collected",
+ defaultScaler.collectMetrics().getMaxObservedRate()
+ );
+
+ // Disabling usePollIdleRatio switches the idle cost to the
processing-rate-based estimate,
+ // which requires tracking the max observed processing rate.
+ CostBasedAutoScalerConfig configWithoutPollIdleRatio =
CostBasedAutoScalerConfig.builder()
+
.taskCountMax(10)
+
.taskCountMin(1)
+
.enableTaskAutoScaler(true)
+
.usePollIdleRatio(false)
+
.build();
+ when(supervisor.getStats()).thenReturn(
+ buildTaskStatsForRate(500.0),
+ buildTaskStatsForRate(9000.0),
+ buildTaskStatsForRate(300.0)
+ );
+ CostBasedAutoScaler autoScalerWithoutPollIdleRatio =
+ new CostBasedAutoScaler(supervisor, configWithoutPollIdleRatio, spec,
emitter);
+
+ Assert.assertEquals(
+ "First sample becomes the watermark",
+ 500.0,
+ autoScalerWithoutPollIdleRatio.collectMetrics().getMaxObservedRate(),
+ 0.0001
+ );
+ Assert.assertEquals(
+ "Watermark tracks the max across observed samples",
+ 9000.0,
+ autoScalerWithoutPollIdleRatio.collectMetrics().getMaxObservedRate(),
+ 0.0001
+ );
+ Assert.assertEquals(
+ "Watermark does not drop when a lower rate is observed",
+ 9000.0,
+ autoScalerWithoutPollIdleRatio.collectMetrics().getMaxObservedRate(),
+ 0.0001
+ );
+ }
+
+ private Map<String, Map<String, Object>> buildTaskStatsForRate(double
processedRate)
+ {
+ Map<String, Map<String, Object>> stats = new HashMap<>();
+ stats.put("0", Collections.singletonMap("task-0",
buildTaskStatsWithMovingAverage(processedRate)));
+ return stats;
+ }
+
private CostMetrics createMetrics(
double avgPartitionLag,
int currentTaskCount,
@@ -576,7 +657,8 @@ public class CostBasedAutoScalerTest
partitionCount,
pollIdleRatio,
3600,
- 1000.0
+ 1000.0,
+ 0.
);
}
@@ -591,6 +673,7 @@ public class CostBasedAutoScalerTest
when(mockSupervisorSpec.getDataSources()).thenReturn(List.of("test-datasource"));
when(mockSupervisor.getIoConfig()).thenReturn(mockIoConfig);
when(mockIoConfig.getStream()).thenReturn("test-stream");
+ when(mockIoConfig.getTaskDuration()).thenReturn(Duration.standardHours(1));
return new CostBasedAutoScaler(mockSupervisor, config, mockSupervisorSpec,
mockEmitter);
}
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/WeightedCostFunctionTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/WeightedCostFunctionTest.java
index 290a61bef00..2c9d462b6f2 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/WeightedCostFunctionTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/WeightedCostFunctionTest.java
@@ -419,6 +419,86 @@ public class WeightedCostFunctionTest
);
}
+ @Test
+ public void testEstimateIdleRatioFromProcessingRate()
+ {
+ // 75% utilization (750/1000) -> idle = 0.25
+ final CostMetrics utilized = createMetricsWithMaxObservedRate(750.0,
1000.0, 0.3);
+ Assert.assertEquals(
+ 0.25,
+ utilized.estimateIdleRatioFromProcessingRate(),
+ 0.0001
+ );
+
+ // Utilization above 100% (rate exceeds the watermark) clamps idle to 0,
not negative
+ final CostMetrics overUtilized = createMetricsWithMaxObservedRate(1500.0,
1000.0, 0.3);
+ Assert.assertEquals(
+ 0.0,
+ overUtilized.estimateIdleRatioFromProcessingRate(),
+ 0.0001
+ );
+
+ // No throughput baseline yet (maxObservedRate=0) -> return negative
(unknown), never NaN from 0/0
+ final CostMetrics noBaseline = createMetricsWithMaxObservedRate(0.0, 0.0,
0.3);
+ Assert.assertTrue(noBaseline.estimateIdleRatioFromProcessingRate() < 0);
+ }
+
+ @Test
+ public void testComputeCostSwitchesBetweenPollIdleRatioAndUtilizationRatio()
+ {
+ CostBasedAutoScalerConfig idleOnlyConfig =
CostBasedAutoScalerConfig.builder()
+
.taskCountMax(100)
+
.taskCountMin(1)
+
.enableTaskAutoScaler(true)
+
.lagWeight(0.0)
+
.idleWeight(1.0)
+
.build();
+ CostBasedAutoScalerConfig utilizationConfig =
CostBasedAutoScalerConfig.builder()
+
.taskCountMax(100)
+
.taskCountMin(1)
+
.enableTaskAutoScaler(true)
+
.lagWeight(0.0)
+
.idleWeight(1.0)
+
.usePollIdleRatio(false)
+
.build();
+
+ // pollIdleRatio says 90% idle, but utilization (100/1000 used) says 90%
idle too -- pick values that
+ // diverge so the two code paths are distinguishable.
+ CostMetrics metrics = createMetricsWithMaxObservedRate(100.0, 1000.0, 0.9);
+
+ double costWithPollIdleRatio = costFunction.computeCost(metrics, 10,
idleOnlyConfig).totalCost();
+ Assert.assertEquals(
+ "Default config should cost using the raw pollIdleRatio",
+ costFunction.uShapedIdleCost(0.9, 10),
+ costWithPollIdleRatio,
+ 0.0001
+ );
+
+ double costWithUtilizationRatio = costFunction.computeCost(metrics, 10,
utilizationConfig).totalCost();
+ Assert.assertEquals(
+ "usePollIdleRatio=false should cost using the rate-derived idle ratio
instead of pollIdleRatio",
+ costFunction.uShapedIdleCost(0.9, 10),
+ costWithUtilizationRatio,
+ 0.0001
+ );
+
+ // Now diverge pollIdleRatio from the utilization-derived value to prove
the flag actually switches sources.
+ CostMetrics divergingMetrics = createMetricsWithMaxObservedRate(100.0,
1000.0, 0.1);
+ double costStillPollIdle = costFunction.computeCost(divergingMetrics, 10,
idleOnlyConfig).totalCost();
+ double costStillUtilization = costFunction.computeCost(divergingMetrics,
10, utilizationConfig).totalCost();
+ Assert.assertEquals(
+ costFunction.uShapedIdleCost(0.1, 10),
+ costStillPollIdle,
+ 0.0001
+ );
+ Assert.assertEquals(
+ "Utilization-derived idle ratio (0.9) should be used instead of the
diverging pollIdleRatio (0.1)",
+ costFunction.uShapedIdleCost(0.9, 10),
+ costStillUtilization,
+ 0.0001
+ );
+ }
+
private CostMetrics createMetrics(
double avgPartitionLag,
int currentTaskCount,
@@ -432,10 +512,20 @@ public class WeightedCostFunctionTest
partitionCount,
pollIdleRatio,
3600,
+ 1000.0,
1000.0
);
}
+ private CostMetrics createMetricsWithMaxObservedRate(
+ double avgProcessingRate,
+ double maxObservedRate,
+ double pollIdleRatio
+ )
+ {
+ return new CostMetrics(0.0, 10, 100, pollIdleRatio, 3600,
avgProcessingRate, maxObservedRate);
+ }
+
private CostMetrics createMetricsWithRate(
double avgPartitionLag,
int currentTaskCount,
@@ -450,7 +540,8 @@ public class WeightedCostFunctionTest
partitionCount,
pollIdleRatio,
3600,
- avgProcessingRate
+ avgProcessingRate,
+ 1000.0
);
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
