From 88add9dea87e197e17f7c9b175f843c9228146c5 Mon Sep 17 00:00:00 2001 From: Kashif Faraz Date: Tue, 30 Jun 2026 13:15:06 +0530 Subject: [PATCH 1/4] Fix metric values of costBased auto-scaler --- .../autoscaler/CostBasedAutoScaler.java | 18 +++++-- .../autoscaler/CostBasedAutoScalerConfig.java | 47 ++++--------------- .../CostBasedAutoScalerConfigTest.java | 32 ++++++------- .../autoscaler/CostBasedAutoScalerTest.java | 2 +- website/.spelling | 2 - 5 files changed, 37 insertions(+), 64 deletions(-) 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 ed3454c929a5..591b0b430808 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 @@ -64,7 +64,9 @@ public class CostBasedAutoScaler implements SupervisorTaskAutoScaler private static final EmittingLogger log = new EmittingLogger(CostBasedAutoScaler.class); public static final String LAG_WEIGHT_METRIC = "task/autoScaler/costBased/lagWeight"; + public static final String LAG_COST_METRIC = "task/autoScaler/costBased/lagCost"; public static final String IDLE_WEIGHT_METRIC = "task/autoScaler/costBased/idleWeight"; + public static final String IDLE_COST_METRIC = "task/autoScaler/costBased/idleCost"; 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"; @@ -295,10 +297,18 @@ int computeOptimalTaskCount(CostMetrics metrics) } 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(LAG_COST_METRIC, optimalCost.lagCost())); + emitter.emit(getMetricBuilder().setMetric(IDLE_COST_METRIC, optimalCost.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 (optimalTaskCount != currentTaskCount) { log.info( 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 167ffcaf9c77..1c01b15f9e1d 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.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 @@ @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; @@ -71,9 +67,6 @@ public class CostBasedAutoScalerConfig implements AutoScalerConfig /** * Creates a new CostBasedAutoScalerConfig instance. - *

- * Note: useTaskCountBoundaries and highLagThreshold are kept for backward compatibility, - * but effectively they are removed. */ @JsonCreator public CostBasedAutoScalerConfig( @@ -81,7 +74,6 @@ public CostBasedAutoScalerConfig( @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, @@ -94,27 +86,15 @@ public CostBasedAutoScalerConfig( @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.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); @@ -194,7 +174,7 @@ public Integer getTaskCountStart() @JsonProperty public long getMinTriggerScaleActionFrequencyMillis() { - return minTriggerScaleActionFrequencyMillis; + return -1; } @Override @@ -288,7 +268,7 @@ public boolean isUsePollIdleRatio() @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 @@ -306,7 +286,6 @@ public boolean equals(Object o) 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 @@ -328,7 +307,6 @@ public int hashCode() taskCountMax, taskCountMin, taskCountStart, - minTriggerScaleActionFrequencyMillis, stopTaskCountRatio, scaleActionPeriodMillis, lagWeight, @@ -350,7 +328,6 @@ public String toString() ", taskCountMax=" + taskCountMax + ", taskCountMin=" + taskCountMin + ", taskCountStart=" + taskCountStart + - ", minTriggerScaleActionFrequencyMillis=" + minTriggerScaleActionFrequencyMillis + ", stopTaskCountRatio=" + stopTaskCountRatio + ", scaleActionPeriodMillis=" + scaleActionPeriodMillis + ", lagWeight=" + lagWeight + @@ -374,7 +351,6 @@ public static class Builder private Integer taskCountMin; private Boolean enableTaskAutoScaler = true; private Integer taskCountStart; - private Long minTriggerScaleActionFrequencyMillis; private Double stopTaskCountRatio; private Long scaleActionPeriodMillis; private Double lagWeight; @@ -414,12 +390,6 @@ public Builder taskCountStart(Integer taskCountStart) return this; } - public Builder minTriggerScaleActionFrequencyMillis(long minTriggerScaleActionFrequencyMillis) - { - this.minTriggerScaleActionFrequencyMillis = minTriggerScaleActionFrequencyMillis; - return this; - } - public Builder stopTaskCountRatio(Double stopTaskCountRatio) { this.stopTaskCountRatio = stopTaskCountRatio; @@ -487,7 +457,6 @@ public CostBasedAutoScalerConfig build() 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 6a8eec3e3d90..7d666a9cb2e7 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 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; @SuppressWarnings("TextBlockMigration") public class CostBasedAutoScalerConfigTest @@ -96,12 +96,12 @@ public void testSerdeWithDefaults() throws Exception 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); // 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()); @@ -220,8 +220,8 @@ public void testScaleDelayDefaults() throws Exception .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() @@ -230,7 +230,7 @@ public void testScaleDelayDefaults() throws Exception .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() @@ -238,7 +238,7 @@ public void testScaleDelayDefaults() throws Exception .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 @@ -261,9 +261,9 @@ public void testScaleDelayDefaults() throws Exception @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. { @@ -285,8 +285,7 @@ public void testMinTriggerScaleActionFrequencyMillisSerdeCompat() throws Excepti + "\"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); } @@ -313,7 +312,6 @@ public void testMinTriggerScaleActionFrequencyMillisSerdeCompat() throws Excepti + "\"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); @@ -328,7 +326,6 @@ public void testMinTriggerScaleActionFrequencyMillisSerdeCompat() throws Excepti + "\"minScaleUpDelay\":\"PT2M\"}", CostBasedAutoScalerConfig.class ); - Assert.assertEquals(900_000L, config.getMinTriggerScaleActionFrequencyMillis()); Assert.assertEquals(Duration.standardMinutes(2), config.getMinScaleUpDelay()); Assert.assertEquals(defaultDown, config.getMinScaleDownDelay()); assertRoundTrips(config); @@ -342,8 +339,7 @@ public void testMinTriggerScaleActionFrequencyMillisSerdeCompat() throws Excepti + "\"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 d773e20bfcb3..51f96173d955 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 @@ -512,7 +512,7 @@ public void testComputeTaskCountForRolloverAndConfigProperties() .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 7452721a1724..36213563c289 100644 --- a/website/.spelling +++ b/website/.spelling @@ -906,8 +906,6 @@ c3.2xlarge defaultManualBrokerService maxPriority minPriority -minScaleUpDelay -minScaleDownDelay NUMBER_FEATURES NUMBER_OF_CONTRIBUTORS PreparedStatement From 8d5c46c609098e30622769eee551340a79346bde Mon Sep 17 00:00:00 2001 From: Kashif Faraz Date: Tue, 30 Jun 2026 13:35:44 +0530 Subject: [PATCH 2/4] Fix embedded test --- .../autoscaler/CostBasedAutoScalerIntegrationTest.java | 5 ----- 1 file changed, 5 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 90f78bec1ef3..9844ca0993ef 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 void test_autoScaler_computesOptimalTaskCountAndProduceScaleDown() .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 void test_autoScaler_computesOptimalTaskCountAndProducesScaleUp() .taskCountMax(50) .taskCountStart(lowInitialTaskCount) .scaleActionPeriodMillis(500) - .minTriggerScaleActionFrequencyMillis(1000) .lagWeight(0.8) .idleWeight(0.2) .build(); @@ -212,7 +210,6 @@ public void test_autoScaler_computesOptimalTaskCountAndProducesScaleUp_withUtili .taskCountMax(50) .taskCountStart(lowInitialTaskCount) .scaleActionPeriodMillis(500) - .minTriggerScaleActionFrequencyMillis(1000) .lagWeight(0.8) .idleWeight(0.2) .usePollIdleRatio(false) @@ -258,7 +255,6 @@ public void test_autoScaler_scalesUpAndDown_withSlowPublish() .taskCountMax(4) .lagWeight(0.5) .idleWeight(0.5) - .minTriggerScaleActionFrequencyMillis(10L) .scaleActionPeriodMillis(10L) .minScaleDownDelay(Duration.standardSeconds(1)) .build(); @@ -336,7 +332,6 @@ void test_scaleDownDuringTaskRollover() .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) From b0f7fe887fb6b3801c2a59ceae5717e17b5885b3 Mon Sep 17 00:00:00 2001 From: Kashif Faraz Date: Wed, 1 Jul 2026 16:16:52 +0530 Subject: [PATCH 3/4] temp stuff --- .../autoscaler/CostBasedAutoScaler.java | 6 +- .../autoscaler/WeightedCostFunction.java | 4 +- .../autoscaler/CostBasedAutoScalerTest.java | 94 ++++++++----------- 3 files changed, 42 insertions(+), 62 deletions(-) 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 591b0b430808..553582d87a8d 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 @@ -235,7 +235,6 @@ int computeOptimalTaskCount(CostMetrics metrics) final int[] validTaskCounts = CostBasedAutoScaler.computeValidTaskCounts( partitionCount, - currentTaskCount, config.getTaskCountMin(), config.getTaskCountMax() ); @@ -325,17 +324,16 @@ int computeOptimalTaskCount(CostMetrics metrics) * 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[]{}; } 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 3e9d7e81b8f4..7ec513c55abb 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 @@ -67,13 +67,13 @@ public class WeightedCostFunction * Penalty magnitude applied when idle ratio is 0 (no safety margin). * Controls the steepness of the U-shape on the under-provisioning side. */ - static final double UNDER_PROVISIONING_PENALTY = 2.0; + static final double UNDER_PROVISIONING_PENALTY = 1.0; /** * Penalty magnitude applied when idle ratio is 1 (fully wasted capacity). * Controls the steepness of the U-shape on the over-provisioning side. */ - static final double OVER_PROVISIONING_PENALTY = 1.0; + static final double OVER_PROVISIONING_PENALTY = 2.0; /** * Computes cost for a given task count using compute time metrics. 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 51f96173d955..435427fa37b2 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 void setUp() } @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 From 8f343d9e34cf19b608f81beb23789343e2d60ddb Mon Sep 17 00:00:00 2001 From: Kashif Faraz Date: Thu, 2 Jul 2026 21:26:39 +0530 Subject: [PATCH 4/4] Add another metric --- .../supervisor/autoscaler/CostBasedAutoScaler.java | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) 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 296e62a8bb97..b76c3fe36852 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 @@ -71,6 +71,7 @@ public class CostBasedAutoScaler implements SupervisorTaskAutoScaler 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 @@ -248,6 +249,7 @@ int computeOptimalTaskCount(CostMetrics metrics) // Start with the current task count as optimal int optimalTaskCount = currentTaskCount; CostResult optimalCost = costFunction.computeCost(metrics, currentTaskCount, config); + final double idleRatioEstimatedFromRate = metrics.estimateIdleRatioFromProcessingRate(); log.info( "Computing optimal taskCount for supervisor[%s] with metrics:" @@ -258,7 +260,7 @@ int computeOptimalTaskCount(CostMetrics metrics) metrics.getAvgPartitionLag(), metrics.getAvgProcessingRate(), metrics.getMaxObservedRate(), - metrics.estimateIdleRatioFromProcessingRate(), + idleRatioEstimatedFromRate, metrics.getPollIdleRatio(), config.getLagWeight(), config.getIdleWeight() @@ -310,6 +312,9 @@ int computeOptimalTaskCount(CostMetrics metrics) 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(