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 9fca068da23 minor: Fix metric values of costBased auto-scaler (#19641)
9fca068da23 is described below
commit 9fca068da23e5518b2ca9daf366daa2331775840
Author: Kashif Faraz <[email protected]>
AuthorDate: Thu Jul 9 15:38:14 2026 +0530
minor: Fix metric values of costBased auto-scaler (#19641)
Changes:
- Add metrics `optimalLagCost`, `optimalIdleCost`
- These metrics would help relate the cost function to physical quantities
- Fix the emitted values of `lagWeight` and `idleWeight`
- Do not emit invalid values for `avgProcessingRate` and `avgPollIdleRatio`
- Remove unused auto-scaler config field `minTriggerScaleActionFrequency`.
Use default constants instead.
- Break up the test method `testComputeValidTaskCounts` into multiple test
cases
---
.../CostBasedAutoScalerIntegrationTest.java | 5 --
.../supervisor/autoscaler/CostBasedAutoScaler.java | 54 ++++++++----
.../autoscaler/CostBasedAutoScalerConfig.java | 47 ++---------
.../autoscaler/CostBasedAutoScalerConfigTest.java | 32 ++++----
.../autoscaler/CostBasedAutoScalerTest.java | 96 +++++++++-------------
website/.spelling | 2 -
6 files changed, 101 insertions(+), 135 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 0f7a8ff3048..7066d6d0df4 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
@@ -108,7 +108,6 @@ public class CostBasedAutoScalerIntegrationTest extends
StreamIndexTestBase
.taskCountMax(100)
.taskCountStart(initialTaskCount)
.scaleActionPeriodMillis(1900)
- .minTriggerScaleActionFrequencyMillis(2000)
// Weight configuration: strongly favor lag reduction over idle time
.lagWeight(0.9)
.idleWeight(0.1)
@@ -163,7 +162,6 @@ public class CostBasedAutoScalerIntegrationTest extends
StreamIndexTestBase
.taskCountMax(50)
.taskCountStart(lowInitialTaskCount)
.scaleActionPeriodMillis(500)
- .minTriggerScaleActionFrequencyMillis(1000)
.lagWeight(0.8)
.idleWeight(0.2)
.build();
@@ -212,7 +210,6 @@ public class CostBasedAutoScalerIntegrationTest extends
StreamIndexTestBase
.taskCountMax(50)
.taskCountStart(lowInitialTaskCount)
.scaleActionPeriodMillis(500)
- .minTriggerScaleActionFrequencyMillis(1000)
.lagWeight(0.8)
.idleWeight(0.2)
.usePollIdleRatio(false)
@@ -258,7 +255,6 @@ public class CostBasedAutoScalerIntegrationTest extends
StreamIndexTestBase
.taskCountMax(4)
.lagWeight(0.5)
.idleWeight(0.5)
- .minTriggerScaleActionFrequencyMillis(10L)
.scaleActionPeriodMillis(10L)
.minScaleDownDelay(Duration.standardSeconds(1))
.build();
@@ -336,7 +332,6 @@ public class CostBasedAutoScalerIntegrationTest extends
StreamIndexTestBase
.taskCountMin(1)
.taskCountMax(10)
.scaleActionPeriodMillis(100)
- .minTriggerScaleActionFrequencyMillis(100)
// High idle weight ensures scale-down when tasks are mostly idle
(little data to process)
.lagWeight(0.1)
.idleWeight(0.9)
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 bf07545aaa5..b089df3bcd1 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
@@ -65,10 +65,15 @@ public class CostBasedAutoScaler implements
SupervisorTaskAutoScaler
public static final String LAG_WEIGHT_METRIC =
"task/autoScaler/costBased/lagWeight";
public static final String IDLE_WEIGHT_METRIC =
"task/autoScaler/costBased/idleWeight";
+ public static final String CURRENT_LAG_COST_METRIC =
"task/autoScaler/costBased/currentLagCost";
+ public static final String CURRENT_IDLE_COST_METRIC =
"task/autoScaler/costBased/currentIdleCost";
+ public static final String OPTIMAL_LAG_COST_METRIC =
"task/autoScaler/costBased/optimalLagCost";
+ public static final String OPTIMAL_IDLE_COST_METRIC =
"task/autoScaler/costBased/optimalIdleCost";
public static final String OPTIMAL_TASK_COUNT_METRIC =
"task/autoScaler/costBased/optimalTaskCount";
public static final String INVALID_METRICS_COUNT =
"task/autoScaler/costBased/invalidMetrics";
public static final String AVG_PROCESSING_RATE_METRIC =
"task/autoScaler/costBased/avgProcessingRate";
public static final String AVG_POLL_IDLE_RATIO =
"task/autoScaler/costBased/avgPollIdleRatio";
+ public static final String IDLE_RATIO_ESTIMATED_FROM_RATE =
"task/autoScaler/costBased/idleRatioFromRate";
/**
* Maximum number of candidate task counts to evaluate above or below the
current task count
@@ -233,7 +238,6 @@ public class CostBasedAutoScaler implements
SupervisorTaskAutoScaler
final int[] validTaskCounts = CostBasedAutoScaler.computeValidTaskCounts(
partitionCount,
- currentTaskCount,
config.getTaskCountMin(),
config.getTaskCountMax()
);
@@ -245,8 +249,10 @@ public class CostBasedAutoScaler implements
SupervisorTaskAutoScaler
}
// Start with the current task count as optimal
+ final CostResult currentCost = costFunction.computeCost(metrics,
currentTaskCount, config);
int optimalTaskCount = currentTaskCount;
- CostResult optimalCost = costFunction.computeCost(metrics,
currentTaskCount, config);
+ CostResult optimalCost = currentCost;
+ final double idleRatioEstimatedFromRate =
metrics.estimateIdleRatioFromProcessingRate();
log.info(
"Computing optimal taskCount for supervisor[%s] with metrics:"
@@ -257,7 +263,7 @@ public class CostBasedAutoScaler implements
SupervisorTaskAutoScaler
metrics.getAvgPartitionLag(),
metrics.getAvgProcessingRate(),
metrics.getMaxObservedRate(),
- metrics.estimateIdleRatioFromProcessingRate(),
+ idleRatioEstimatedFromRate,
metrics.getPollIdleRatio(),
config.getLagWeight(),
config.getIdleWeight()
@@ -297,16 +303,29 @@ public class CostBasedAutoScaler implements
SupervisorTaskAutoScaler
}
emitter.emit(getMetricBuilder().setMetric(OPTIMAL_TASK_COUNT_METRIC,
(long) optimalTaskCount));
- emitter.emit(getMetricBuilder().setMetric(LAG_WEIGHT_METRIC,
optimalCost.lagCost()));
- emitter.emit(getMetricBuilder().setMetric(IDLE_WEIGHT_METRIC,
optimalCost.idleCost()));
- emitter.emit(getMetricBuilder().setMetric(AVG_PROCESSING_RATE_METRIC,
metrics.getAvgProcessingRate()));
- emitter.emit(getMetricBuilder().setMetric(AVG_POLL_IDLE_RATIO,
metrics.getPollIdleRatio()));
+ emitter.emit(getMetricBuilder().setMetric(LAG_WEIGHT_METRIC,
config.getLagWeight()));
+ emitter.emit(getMetricBuilder().setMetric(IDLE_WEIGHT_METRIC,
config.getIdleWeight()));
+ emitter.emit(getMetricBuilder().setMetric(CURRENT_LAG_COST_METRIC,
currentCost.lagCost()));
+ emitter.emit(getMetricBuilder().setMetric(CURRENT_IDLE_COST_METRIC,
currentCost.idleCost()));
+
+ // Emit avg rate and idle metrics only if they are available
+ if (metrics.getAvgProcessingRate() >= 0) {
+ emitter.emit(getMetricBuilder().setMetric(AVG_PROCESSING_RATE_METRIC,
metrics.getAvgProcessingRate()));
+ }
+ if (metrics.getPollIdleRatio() >= 0) {
+ emitter.emit(getMetricBuilder().setMetric(AVG_POLL_IDLE_RATIO,
metrics.getPollIdleRatio()));
+ }
+ if (idleRatioEstimatedFromRate >= 0) {
+
emitter.emit(getMetricBuilder().setMetric(IDLE_RATIO_ESTIMATED_FROM_RATE,
idleRatioEstimatedFromRate));
+ }
if (optimalTaskCount != currentTaskCount) {
log.info(
"Optimal taskCount[%d] for supervisor[%s] has lowest cost[%.4f] out
of the following candidates: %n%s",
optimalTaskCount, supervisorId, optimalCost.totalCost(),
constructCostTable(validTaskCounts, costResults)
);
+ emitter.emit(getMetricBuilder().setMetric(OPTIMAL_LAG_COST_METRIC,
optimalCost.lagCost()));
+ emitter.emit(getMetricBuilder().setMetric(OPTIMAL_IDLE_COST_METRIC,
optimalCost.idleCost()));
}
// Scale-up is applied eagerly; scale-down may be deferred by
computeTaskCountForScaleAction().
@@ -317,17 +336,16 @@ public class CostBasedAutoScaler implements
SupervisorTaskAutoScaler
* Generates valid task counts by converting every possible
partitions-per-task ratio
* into a task count and filtering by configured min/max task count bounds.
*
- * @return list of valid task counts within bounds
+ * @return array of valid task counts within bounds
*/
@SuppressWarnings({"ReassignedVariable"})
static int[] computeValidTaskCounts(
int partitionCount,
- int currentTaskCount,
int taskCountMin,
int taskCountMax
)
{
- if (partitionCount <= 0 || currentTaskCount <= 0) {
+ if (partitionCount <= 0) {
return new int[]{};
}
@@ -368,8 +386,12 @@ public class CostBasedAutoScaler implements
SupervisorTaskAutoScaler
if (autoScalerMetricsMap instanceof Map) {
Object pollIdleRatioAvg = ((Map<?, ?>)
autoScalerMetricsMap).get(SeekableStreamIndexTaskRunner.POLL_IDLE_RATIO_KEY);
if (pollIdleRatioAvg instanceof Number) {
- sum += ((Number) pollIdleRatioAvg).doubleValue();
- count++;
+ double taskPollIdleRatio = ((Number)
pollIdleRatioAvg).doubleValue();
+ // Do not include negative values
+ if (taskPollIdleRatio >= 0) {
+ sum += taskPollIdleRatio;
+ count++;
+ }
}
}
}
@@ -414,8 +436,12 @@ public class CostBasedAutoScaler implements
SupervisorTaskAutoScaler
if (movingAvgObj instanceof Map) {
Object processedRate = ((Map<?, ?>)
movingAvgObj).get(RowIngestionMeters.PROCESSED);
if (processedRate instanceof Number) {
- sum += ((Number) processedRate).doubleValue();
- count++;
+ // Do not include negative values
+ double taskProcessedRate = ((Number)
processedRate).doubleValue();
+ if (taskProcessedRate >= 0) {
+ sum += taskProcessedRate;
+ count++;
+ }
}
}
}
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 fa4d547eae4..785fa3257b5 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
@@ -28,7 +28,6 @@ import
org.apache.druid.indexing.overlord.supervisor.Supervisor;
import org.apache.druid.indexing.overlord.supervisor.SupervisorSpec;
import
org.apache.druid.indexing.overlord.supervisor.autoscaler.SupervisorTaskAutoScaler;
import
org.apache.druid.indexing.seekablestream.supervisor.SeekableStreamSupervisor;
-import org.apache.druid.java.util.emitter.EmittingLogger;
import org.apache.druid.java.util.emitter.service.ServiceEmitter;
import org.joda.time.Duration;
@@ -45,18 +44,15 @@ import java.util.Objects;
@JsonInclude(JsonInclude.Include.NON_NULL)
public class CostBasedAutoScalerConfig implements AutoScalerConfig
{
- private static final EmittingLogger LOG = new
EmittingLogger(CostBasedAutoScalerConfig.class);
-
- static final long DEFAULT_SCALE_ACTION_PERIOD_MILLIS = 10 * 60 * 1000; // 10
minutes
static final double DEFAULT_LAG_WEIGHT = 0.4;
static final double DEFAULT_IDLE_WEIGHT = 0.6;
- static final Duration DEFAULT_MIN_SCALE_DELAY =
Duration.millis(DEFAULT_SCALE_ACTION_PERIOD_MILLIS * 3);
+ static final Duration DEFAULT_MIN_SCALE_UP_DELAY =
Duration.standardMinutes(10);
+ static final Duration DEFAULT_MIN_SCALE_DOWN_DELAY =
Duration.standardMinutes(30);
private final boolean enableTaskAutoScaler;
private final int taskCountMax;
private final int taskCountMin;
private final Integer taskCountStart;
- private final long minTriggerScaleActionFrequencyMillis;
private final Double stopTaskCountRatio;
private final long scaleActionPeriodMillis;
@@ -72,9 +68,6 @@ public class CostBasedAutoScalerConfig implements
AutoScalerConfig
/**
* Creates a new CostBasedAutoScalerConfig instance.
- * <p>
- * Note: useTaskCountBoundaries and highLagThreshold are kept for backward
compatibility,
- * but effectively they are removed.
*/
@JsonCreator
public CostBasedAutoScalerConfig(
@@ -82,7 +75,6 @@ public class CostBasedAutoScalerConfig implements
AutoScalerConfig
@JsonProperty("taskCountMin") Integer taskCountMin,
@Nullable @JsonProperty("enableTaskAutoScaler") Boolean
enableTaskAutoScaler,
@Nullable @JsonProperty("taskCountStart") Integer taskCountStart,
- @Nullable @JsonProperty("minTriggerScaleActionFrequencyMillis") Long
minTriggerScaleActionFrequencyMillis,
@Nullable @JsonProperty("stopTaskCountRatio") Double stopTaskCountRatio,
@Nullable @JsonProperty("scaleActionPeriodMillis") Long
scaleActionPeriodMillis,
@Nullable @JsonProperty("lagWeight") Double lagWeight,
@@ -96,18 +88,9 @@ public class CostBasedAutoScalerConfig implements
AutoScalerConfig
@Nullable @JsonProperty("usePollIdleRatio") Boolean usePollIdleRatio
)
{
- this.enableTaskAutoScaler = enableTaskAutoScaler != null ?
enableTaskAutoScaler : false;
-
- // Timing configuration with defaults
- this.scaleActionPeriodMillis = scaleActionPeriodMillis != null
- ? scaleActionPeriodMillis
- : DEFAULT_SCALE_ACTION_PERIOD_MILLIS;
- this.minTriggerScaleActionFrequencyMillis = Configs.valueOrDefault(
- minTriggerScaleActionFrequencyMillis,
- DEFAULT_SCALE_ACTION_PERIOD_MILLIS
- );
+ this.enableTaskAutoScaler = Configs.valueOrDefault(enableTaskAutoScaler,
false);
+ this.scaleActionPeriodMillis =
Configs.valueOrDefault(scaleActionPeriodMillis,
DEFAULT_MIN_SCALE_UP_DELAY.getMillis());
- // Cost function weights with defaults
this.lagWeight = Configs.valueOrDefault(lagWeight, DEFAULT_LAG_WEIGHT);
this.idleWeight = Configs.valueOrDefault(idleWeight, DEFAULT_IDLE_WEIGHT);
this.optimalTaskIdleRatio = Configs.valueOrDefault(
@@ -116,11 +99,8 @@ public class CostBasedAutoScalerConfig implements
AutoScalerConfig
);
this.useTaskCountBoundariesOnScaleUp =
Configs.valueOrDefault(useTaskCountBoundariesOnScaleUp, false);
this.useTaskCountBoundariesOnScaleDown =
Configs.valueOrDefault(useTaskCountBoundariesOnScaleDown, true);
- this.minScaleUpDelay = Configs.valueOrDefault(
- minScaleUpDelay,
- Duration.millis(this.minTriggerScaleActionFrequencyMillis)
- );
- this.minScaleDownDelay = Configs.valueOrDefault(minScaleDownDelay,
DEFAULT_MIN_SCALE_DELAY);
+ this.minScaleUpDelay = Configs.valueOrDefault(minScaleUpDelay,
DEFAULT_MIN_SCALE_UP_DELAY);
+ this.minScaleDownDelay = Configs.valueOrDefault(minScaleDownDelay,
DEFAULT_MIN_SCALE_DOWN_DELAY);
this.scaleDownDuringTaskRolloverOnly =
Configs.valueOrDefault(scaleDownDuringTaskRolloverOnly, false);
this.usePollIdleRatio = Configs.valueOrDefault(usePollIdleRatio, true);
@@ -204,7 +184,7 @@ public class CostBasedAutoScalerConfig implements
AutoScalerConfig
@JsonProperty
public long getMinTriggerScaleActionFrequencyMillis()
{
- return minTriggerScaleActionFrequencyMillis;
+ return -1;
}
@Override
@@ -308,7 +288,7 @@ public class CostBasedAutoScalerConfig implements
AutoScalerConfig
@Override
public SupervisorTaskAutoScaler createAutoScaler(Supervisor supervisor,
SupervisorSpec spec, ServiceEmitter emitter)
{
- return new CostBasedAutoScaler((SeekableStreamSupervisor) supervisor,
this, spec, emitter);
+ return new CostBasedAutoScaler((SeekableStreamSupervisor<?, ?, ?>)
supervisor, this, spec, emitter);
}
@Override
@@ -326,7 +306,6 @@ public class CostBasedAutoScalerConfig implements
AutoScalerConfig
return enableTaskAutoScaler == that.enableTaskAutoScaler
&& taskCountMax == that.taskCountMax
&& taskCountMin == that.taskCountMin
- && minTriggerScaleActionFrequencyMillis ==
that.minTriggerScaleActionFrequencyMillis
&& scaleActionPeriodMillis == that.scaleActionPeriodMillis
&& Double.compare(that.lagWeight, lagWeight) == 0
&& Double.compare(that.idleWeight, idleWeight) == 0
@@ -349,7 +328,6 @@ public class CostBasedAutoScalerConfig implements
AutoScalerConfig
taskCountMax,
taskCountMin,
taskCountStart,
- minTriggerScaleActionFrequencyMillis,
stopTaskCountRatio,
scaleActionPeriodMillis,
lagWeight,
@@ -372,7 +350,6 @@ public class CostBasedAutoScalerConfig implements
AutoScalerConfig
", taskCountMax=" + taskCountMax +
", taskCountMin=" + taskCountMin +
", taskCountStart=" + taskCountStart +
- ", minTriggerScaleActionFrequencyMillis=" +
minTriggerScaleActionFrequencyMillis +
", stopTaskCountRatio=" + stopTaskCountRatio +
", scaleActionPeriodMillis=" + scaleActionPeriodMillis +
", lagWeight=" + lagWeight +
@@ -397,7 +374,6 @@ public class CostBasedAutoScalerConfig implements
AutoScalerConfig
private Integer taskCountMin;
private Boolean enableTaskAutoScaler = true;
private Integer taskCountStart;
- private Long minTriggerScaleActionFrequencyMillis;
private Double stopTaskCountRatio;
private Long scaleActionPeriodMillis;
private Double lagWeight;
@@ -438,12 +414,6 @@ public class CostBasedAutoScalerConfig implements
AutoScalerConfig
return this;
}
- public Builder minTriggerScaleActionFrequencyMillis(long
minTriggerScaleActionFrequencyMillis)
- {
- this.minTriggerScaleActionFrequencyMillis =
minTriggerScaleActionFrequencyMillis;
- return this;
- }
-
public Builder stopTaskCountRatio(Double stopTaskCountRatio)
{
this.stopTaskCountRatio = stopTaskCountRatio;
@@ -517,7 +487,6 @@ public class CostBasedAutoScalerConfig implements
AutoScalerConfig
taskCountMin,
enableTaskAutoScaler,
taskCountStart,
- minTriggerScaleActionFrequencyMillis,
stopTaskCountRatio,
scaleActionPeriodMillis,
lagWeight,
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 6d09221b410..45771d782ab 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
@@ -27,8 +27,8 @@ import org.junit.Test;
import static
org.apache.druid.indexing.seekablestream.supervisor.autoscaler.CostBasedAutoScalerConfig.DEFAULT_IDLE_WEIGHT;
import static
org.apache.druid.indexing.seekablestream.supervisor.autoscaler.CostBasedAutoScalerConfig.DEFAULT_LAG_WEIGHT;
-import static
org.apache.druid.indexing.seekablestream.supervisor.autoscaler.CostBasedAutoScalerConfig.DEFAULT_MIN_SCALE_DELAY;
-import static
org.apache.druid.indexing.seekablestream.supervisor.autoscaler.CostBasedAutoScalerConfig.DEFAULT_SCALE_ACTION_PERIOD_MILLIS;
+import static
org.apache.druid.indexing.seekablestream.supervisor.autoscaler.CostBasedAutoScalerConfig.DEFAULT_MIN_SCALE_DOWN_DELAY;
+import static
org.apache.druid.indexing.seekablestream.supervisor.autoscaler.CostBasedAutoScalerConfig.DEFAULT_MIN_SCALE_UP_DELAY;
import static
org.apache.druid.indexing.seekablestream.supervisor.autoscaler.WeightedCostFunction.OPTIMAL_TASK_IDLE_RATIO;
@SuppressWarnings("TextBlockMigration")
@@ -99,13 +99,13 @@ public class CostBasedAutoScalerConfigTest
Assert.assertEquals(2, config.getTaskCountMin());
// Check defaults
- Assert.assertEquals(DEFAULT_SCALE_ACTION_PERIOD_MILLIS,
config.getScaleActionPeriodMillis());
+ Assert.assertEquals(DEFAULT_MIN_SCALE_UP_DELAY.getMillis(),
config.getScaleActionPeriodMillis());
Assert.assertEquals(DEFAULT_LAG_WEIGHT, config.getLagWeight(), 0.001);
Assert.assertEquals(DEFAULT_IDLE_WEIGHT, config.getIdleWeight(), 0.001);
Assert.assertEquals(OPTIMAL_TASK_IDLE_RATIO,
config.getOptimalTaskIdleRatio(), 0.001);
// minScaleUpDelay and minScaleDownDelay each have their own independent
default
- Assert.assertEquals(Duration.millis(DEFAULT_SCALE_ACTION_PERIOD_MILLIS),
config.getMinScaleUpDelay());
- Assert.assertEquals(DEFAULT_MIN_SCALE_DELAY,
config.getMinScaleDownDelay());
+ Assert.assertEquals(DEFAULT_MIN_SCALE_UP_DELAY,
config.getMinScaleUpDelay());
+ Assert.assertEquals(DEFAULT_MIN_SCALE_DOWN_DELAY,
config.getMinScaleDownDelay());
Assert.assertFalse(config.isScaleDownOnTaskRolloverOnly());
Assert.assertTrue(config.isUsePollIdleRatio());
Assert.assertFalse(config.isUseTaskCountBoundariesOnScaleUp());
@@ -248,8 +248,8 @@ public class CostBasedAutoScalerConfigTest
.taskCountMax(10)
.taskCountMin(1)
.build();
- Assert.assertEquals(Duration.millis(DEFAULT_SCALE_ACTION_PERIOD_MILLIS),
defaults.getMinScaleUpDelay());
- Assert.assertEquals(DEFAULT_MIN_SCALE_DELAY,
defaults.getMinScaleDownDelay());
+ Assert.assertEquals(DEFAULT_MIN_SCALE_UP_DELAY,
defaults.getMinScaleUpDelay());
+ Assert.assertEquals(DEFAULT_MIN_SCALE_DOWN_DELAY,
defaults.getMinScaleDownDelay());
// Only minScaleUpDelay set: up uses explicit value, down uses its default
CostBasedAutoScalerConfig upOnly = CostBasedAutoScalerConfig.builder()
@@ -258,7 +258,7 @@ public class CostBasedAutoScalerConfigTest
.minScaleUpDelay(Duration.standardMinutes(5))
.build();
Assert.assertEquals(Duration.standardMinutes(5),
upOnly.getMinScaleUpDelay());
- Assert.assertEquals(DEFAULT_MIN_SCALE_DELAY,
upOnly.getMinScaleDownDelay());
+ Assert.assertEquals(DEFAULT_MIN_SCALE_DOWN_DELAY,
upOnly.getMinScaleDownDelay());
// Only minScaleDownDelay set: down uses explicit value, up uses its own
default (does not fall back to down)
CostBasedAutoScalerConfig downOnly = CostBasedAutoScalerConfig.builder()
@@ -266,7 +266,7 @@ public class CostBasedAutoScalerConfigTest
.taskCountMin(1)
.minScaleDownDelay(Duration.standardMinutes(20))
.build();
- Assert.assertEquals(Duration.millis(DEFAULT_SCALE_ACTION_PERIOD_MILLIS),
downOnly.getMinScaleUpDelay());
+ Assert.assertEquals(DEFAULT_MIN_SCALE_UP_DELAY,
downOnly.getMinScaleUpDelay());
Assert.assertEquals(Duration.standardMinutes(20),
downOnly.getMinScaleDownDelay());
// Both set: serde roundtrip preserves values
@@ -289,9 +289,9 @@ public class CostBasedAutoScalerConfigTest
@SuppressWarnings("deprecation")
public void testMinTriggerScaleActionFrequencyMillisSerdeCompat() throws
Exception
{
- final long defaultMinTriggerMillis = DEFAULT_SCALE_ACTION_PERIOD_MILLIS;
- final Duration defaultUp =
Duration.millis(DEFAULT_SCALE_ACTION_PERIOD_MILLIS);
- final Duration defaultDown = DEFAULT_MIN_SCALE_DELAY;
+ final long defaultMinTriggerMillis = -1;
+ final Duration defaultUp = DEFAULT_MIN_SCALE_UP_DELAY;
+ final Duration defaultDown = DEFAULT_MIN_SCALE_DOWN_DELAY;
// Backwards-compat: nothing set -> everything uses its own default.
{
@@ -313,8 +313,7 @@ public class CostBasedAutoScalerConfigTest
+ "\"minTriggerScaleActionFrequencyMillis\":900000}",
CostBasedAutoScalerConfig.class
);
- Assert.assertEquals(900_000L,
config.getMinTriggerScaleActionFrequencyMillis());
- Assert.assertEquals(Duration.millis(900_000L),
config.getMinScaleUpDelay());
+ Assert.assertEquals(defaultUp, config.getMinScaleUpDelay());
Assert.assertEquals(defaultDown, config.getMinScaleDownDelay());
assertRoundTrips(config);
}
@@ -341,7 +340,6 @@ public class CostBasedAutoScalerConfigTest
+ "\"minScaleUpDelay\":\"PT2M\",\"minScaleDownDelay\":\"PT15M\"}",
CostBasedAutoScalerConfig.class
);
- Assert.assertEquals(900_000L,
config.getMinTriggerScaleActionFrequencyMillis());
Assert.assertEquals(Duration.standardMinutes(2),
config.getMinScaleUpDelay());
Assert.assertEquals(Duration.standardMinutes(15),
config.getMinScaleDownDelay());
assertRoundTrips(config);
@@ -356,7 +354,6 @@ public class CostBasedAutoScalerConfigTest
+ "\"minScaleUpDelay\":\"PT2M\"}",
CostBasedAutoScalerConfig.class
);
- Assert.assertEquals(900_000L,
config.getMinTriggerScaleActionFrequencyMillis());
Assert.assertEquals(Duration.standardMinutes(2),
config.getMinScaleUpDelay());
Assert.assertEquals(defaultDown, config.getMinScaleDownDelay());
assertRoundTrips(config);
@@ -370,8 +367,7 @@ public class CostBasedAutoScalerConfigTest
+ "\"minScaleDownDelay\":\"PT15M\"}",
CostBasedAutoScalerConfig.class
);
- Assert.assertEquals(900_000L,
config.getMinTriggerScaleActionFrequencyMillis());
- Assert.assertEquals(Duration.millis(900_000L),
config.getMinScaleUpDelay());
+ Assert.assertEquals(defaultUp, config.getMinScaleUpDelay());
Assert.assertEquals(Duration.standardMinutes(15),
config.getMinScaleDownDelay());
assertRoundTrips(config);
}
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 d773e20bfcb..435427fa37b 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
@@ -76,69 +76,51 @@ public class CostBasedAutoScalerTest
}
@Test
- public void testComputeValidTaskCounts()
+ public void test_computeValidTaskCounts_returnsSortedArray()
{
- // For 100 partitions at 25 tasks (4 partitions/task), valid counts
include 25 and 34
- final int[] validTaskCounts = computeValidTaskCounts(
- 100,
- 25,
- 1,
- 100
+ final int partitionCount = 100;
+ final int minTaskCount = 1;
+ final int maxTaskCount = 100;
+ Assert.assertArrayEquals(
+ new int[]{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 12, 13, 15, 17, 20, 25, 34,
50, 100},
+ computeValidTaskCounts(partitionCount, minTaskCount, maxTaskCount)
);
- Assert.assertTrue("Expected current task count to be included",
contains(validTaskCounts, 25));
- Assert.assertTrue("Expected next scale-up option (34) to be included",
contains(validTaskCounts, 34));
+ }
- // Single partition
- final int[] singlePartition = computeValidTaskCounts(
- 1,
- 1,
- 1,
- 100
+ @Test
+ public void test_computeValidTaskCounts_withSinglePartition()
+ {
+ final int partitionCount = 1;
+ final int minTaskCount = 1;
+ final int maxTaskCount = 100;
+ Assert.assertArrayEquals(
+ new int[]{1},
+ computeValidTaskCounts(partitionCount, minTaskCount, maxTaskCount)
);
- Assert.assertTrue("Single partition should yield at least one valid
count", singlePartition.length > 0);
- Assert.assertTrue("Single partition should include task count 1",
contains(singlePartition, 1));
+ }
- // Current exceeds partitions - should still yield valid, deduplicated
options
- final int[] exceedsPartitions = computeValidTaskCounts(
- 2,
- 5,
- 1,
- 100
+ @Test
+ public void test_computeValidTaskCounts_filtersByTaskCountMax()
+ {
+ final int partitionCount = 30;
+ final int taskCountMin = 1;
+ final int taskCountMax = 3;
+ Assert.assertArrayEquals(
+ new int[]{1, 2, 3},
+ computeValidTaskCounts(partitionCount, taskCountMin, taskCountMax)
);
- Assert.assertEquals(2, exceedsPartitions.length);
- Assert.assertTrue(contains(exceedsPartitions, 1));
- Assert.assertTrue(contains(exceedsPartitions, 2));
-
- // Unbounded candidate generation includes both nearby and maximum task
counts.
- final int[] taskCounts = computeValidTaskCounts(30, 3, 1, 30);
- Assert.assertTrue("Valid task counts should include max task count",
contains(taskCounts, 30));
- Assert.assertTrue("Valid task counts should include nearby scale-up task
count", contains(taskCounts, 4));
+ }
- // Respects taskCountMax
- final int[] cappedCounts = computeValidTaskCounts(
- 30,
- 4,
- 1,
- 3
- );
- Assert.assertTrue("Should include taskCountMax when within bounds",
contains(cappedCounts, 3));
- Assert.assertFalse("Should not exceed taskCountMax",
contains(cappedCounts, 4));
-
- // Respects taskCountMin - filters out values below the minimum
- // With partitionCount=100, currentTaskCount=10, the computed range
includes values like 8, 9, 10, 12, 13
- final int[] minCappedCounts = computeValidTaskCounts(100, 10, 10, 100);
- Assert.assertFalse("Should not include values below taskCountMin (8)",
contains(minCappedCounts, 8));
- Assert.assertFalse("Should not include values below taskCountMin (9)",
contains(minCappedCounts, 9));
- Assert.assertTrue("Should include values at taskCountMin (10)",
contains(minCappedCounts, 10));
- Assert.assertTrue("Should include values above taskCountMin (12)",
contains(minCappedCounts, 12));
-
- // Both bounds applied together
- final int[] bothBounds = computeValidTaskCounts(100, 10, 10, 12);
- Assert.assertFalse("Should not include values below taskCountMin (8)",
contains(bothBounds, 8));
- Assert.assertFalse("Should not include values below taskCountMin (9)",
contains(bothBounds, 9));
- Assert.assertFalse("Should not include values above taskCountMax (13)",
contains(bothBounds, 13));
- Assert.assertTrue("Should include values at taskCountMin (10)",
contains(bothBounds, 10));
- Assert.assertTrue("Should include values at taskCountMax (12)",
contains(bothBounds, 12));
+ @Test
+ public void
test_computeValidTaskCounts_filtersByTaskCountMaxAndTaskCountMin()
+ {
+ final int partitionCount = 100;
+ final int taskCountMin = 10;
+ final int taskCountMax = 30;
+ Assert.assertArrayEquals(
+ new int[]{10, 12, 13, 15, 17, 20, 25},
+ computeValidTaskCounts(partitionCount, taskCountMin, taskCountMax)
+ );
}
@Test
@@ -512,7 +494,7 @@ public class CostBasedAutoScalerTest
.enableTaskAutoScaler(true)
.build();
Assert.assertEquals(
- CostBasedAutoScalerConfig.DEFAULT_MIN_SCALE_DELAY,
+ CostBasedAutoScalerConfig.DEFAULT_MIN_SCALE_DOWN_DELAY,
cfgWithDefaults.getMinScaleDownDelay()
);
Assert.assertFalse(cfgWithDefaults.isScaleDownOnTaskRolloverOnly());
diff --git a/website/.spelling b/website/.spelling
index 468df62684f..e1723b254dc 100644
--- a/website/.spelling
+++ b/website/.spelling
@@ -907,8 +907,6 @@ c3.2xlarge
defaultManualBrokerService
maxPriority
minPriority
-minScaleUpDelay
-minScaleDownDelay
NUMBER_FEATURES
NUMBER_OF_CONTRIBUTORS
PreparedStatement
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
