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 0f7a8ff30482..7066d6d0df4e 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) 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 bf07545aaa53..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 @@ -64,11 +64,14 @@ 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"; 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 +236,6 @@ int computeOptimalTaskCount(CostMetrics metrics) final int[] validTaskCounts = CostBasedAutoScaler.computeValidTaskCounts( partitionCount, - currentTaskCount, config.getTaskCountMin(), config.getTaskCountMax() ); @@ -247,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:" @@ -257,7 +260,7 @@ int computeOptimalTaskCount(CostMetrics metrics) metrics.getAvgPartitionLag(), metrics.getAvgProcessingRate(), metrics.getMaxObservedRate(), - metrics.estimateIdleRatioFromProcessingRate(), + idleRatioEstimatedFromRate, metrics.getPollIdleRatio(), config.getLagWeight(), config.getIdleWeight() @@ -297,10 +300,21 @@ 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 (idleRatioEstimatedFromRate >= 0) { + emitter.emit(getMetricBuilder().setMetric(IDLE_RATIO_ESTIMATED_FROM_RATE, idleRatioEstimatedFromRate)); + } if (optimalTaskCount != currentTaskCount) { log.info( @@ -317,17 +331,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/CostBasedAutoScalerConfig.java b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScalerConfig.java index fa4d547eae49..785fa3257b5c 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; @@ -72,9 +68,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( @@ -82,7 +75,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, @@ -96,18 +88,9 @@ 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.optimalTaskIdleRatio = Configs.valueOrDefault( @@ -116,11 +99,8 @@ public CostBasedAutoScalerConfig( ); 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 Integer getTaskCountStart() @JsonProperty public long getMinTriggerScaleActionFrequencyMillis() { - return minTriggerScaleActionFrequencyMillis; + return -1; } @Override @@ -308,7 +288,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 @@ -326,7 +306,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 @@ -349,7 +328,6 @@ public int hashCode() taskCountMax, taskCountMin, taskCountStart, - minTriggerScaleActionFrequencyMillis, stopTaskCountRatio, scaleActionPeriodMillis, lagWeight, @@ -372,7 +350,6 @@ public String toString() ", taskCountMax=" + taskCountMax + ", taskCountMin=" + taskCountMin + ", taskCountStart=" + taskCountStart + - ", minTriggerScaleActionFrequencyMillis=" + minTriggerScaleActionFrequencyMillis + ", stopTaskCountRatio=" + stopTaskCountRatio + ", scaleActionPeriodMillis=" + scaleActionPeriodMillis + ", lagWeight=" + lagWeight + @@ -397,7 +374,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; @@ -438,12 +414,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; @@ -517,7 +487,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 6d09221b410a..45771d782abb 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; import static org.apache.druid.indexing.seekablestream.supervisor.autoscaler.WeightedCostFunction.OPTIMAL_TASK_IDLE_RATIO; @SuppressWarnings("TextBlockMigration") @@ -99,13 +99,13 @@ 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); 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 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() @@ -258,7 +258,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() @@ -266,7 +266,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 @@ -289,9 +289,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. { @@ -313,8 +313,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); } @@ -341,7 +340,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); @@ -356,7 +354,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); @@ -370,8 +367,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..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 @@ -512,7 +494,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