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 a7bcc37897c feat: Introduce lag emergency option for cost-based
autoscaler (#19655)
a7bcc37897c is described below
commit a7bcc37897c72dec3213aaed3a811682c7c9f04e
Author: Sasha Syrotenko <[email protected]>
AuthorDate: Wed Jul 29 19:14:25 2026 +0300
feat: Introduce lag emergency option for cost-based autoscaler (#19655)
---
docs/ingestion/supervisor.md | 14 +-
.../supervisor/autoscaler/CostBasedAutoScaler.java | 70 ++++++--
.../autoscaler/CostBasedAutoScalerConfig.java | 83 ++++++++-
.../supervisor/autoscaler/CostMetrics.java | 7 +-
.../autoscaler/WeightedCostFunction.java | 38 +++-
.../autoscaler/CostBasedAutoScalerConfigTest.java | 60 ++++++-
.../autoscaler/CostBasedAutoScalerMockTest.java | 1 +
.../autoscaler/CostBasedAutoScalerTest.java | 195 ++++++++++++++++++++-
.../autoscaler/WeightedCostFunctionTest.java | 90 ++++++++--
9 files changed, 510 insertions(+), 48 deletions(-)
diff --git a/docs/ingestion/supervisor.md b/docs/ingestion/supervisor.md
index 2a4f2a26006..73445a4890d 100644
--- a/docs/ingestion/supervisor.md
+++ b/docs/ingestion/supervisor.md
@@ -83,7 +83,7 @@ The following table outlines the configuration properties for
`autoScalerConfig`
|`minScaleUpDelay`|Minimum cooldown duration between scale-up actions,
specified as an ISO-8601 duration string. Falls back to
`minTriggerScaleActionFrequencyMillis` if not set.|No||
|`minScaleDownDelay`|Minimum cooldown duration between scale-down actions,
specified as an ISO-8601 duration string. Falls back to
`minTriggerScaleActionFrequencyMillis` if not set.|No||
|`minTriggerScaleActionFrequencyMillis`|**Deprecated.** Use `minScaleUpDelay`
and `minScaleDownDelay` instead. Minimum time interval in milliseconds between
scale actions, used as the fallback when the Duration-based fields are not
set.|No|600000|
-|`autoScalerStrategy`|The algorithm of autoscaler. Druid only supports the
`lagBased` strategy. See [Autoscaler strategy](#autoscaler-strategy) for more
information.|No|`lagBased`|
+|`autoScalerStrategy`|The autoscaler algorithm. Druid supports the `lagBased`
and `costBased` strategies. See [Autoscaler strategy](#autoscaler-strategy) for
more information.|No|`lagBased`|
|`stopTaskCountRatio`|A variable version of `ioConfig.stopTaskCount` with a
valid range of (0.0, 1.0]. Allows the maximum number of stoppable tasks in
steady state to be proportional to the number of tasks currently running.|No||
##### Autoscaler strategy
@@ -206,16 +206,20 @@ At every evaluation interval, Druid computes the score
for each candidate task c
The following table outlines the configuration properties related to the
`costBased` autoscaler strategy:
+The cost-based autoscaler uses aggregate lag, which is the sum of the lag
across all partitions. For Kafka and Rabbit, lag is measured in records. For
Kinesis, lag is measured in milliseconds. Set `criticalLagThreshold` in the
units reported by the stream type.
+
| Property | Description | Required | Default |
|----------|-------------|----------|---------------------------|
-|`scaleActionPeriodMillis`|How often, in milliseconds, Druid evaluates whether
to scale.|No| `600000` (10 min) |
+|`scaleActionPeriodMillis`|How often, in milliseconds, Druid evaluates whether
to scale.|No| `120000` (2 min) |
|`lagWeight`|How much weight to give the lag cost relative to the idle cost.
Higher values make the autoscaler more aggressive about adding tasks to drain
backlog.|No| `0.4` |
|`idleWeight`|How much weight to give the idle cost relative to the lag cost.
Higher values make the autoscaler more aggressive about removing
over-provisioned tasks.|No| `0.6` |
|`useTaskCountBoundariesOnScaleUp`|Limits scale-up to a small step relative to
the current task count, preventing large jumps. Disable to allow the autoscaler
to jump directly to any task count.|No| `false` |
|`useTaskCountBoundariesOnScaleDown`|Limits scale-down to a small step
relative to the current task count, preventing large drops. Disable to allow
the autoscaler to drop directly to any task count.|No| `true`
|
-|`minScaleUpDelay`|Minimum cooldown after a scale-up before the next scale-up
is allowed. Specified as an ISO-8601 duration.|No| `scaleActionPeriodMillis` |
+|`minScaleUpDelay`|Minimum cooldown after a scale-up before the next scale-up
is allowed. Specified as an ISO-8601 duration.|No| `PT15M` |
|`minScaleDownDelay`|Minimum cooldown after a scale-down before the next
scale-down is allowed. Specified as an ISO-8601 duration.|No| `PT30M`
|
|`scaleDownDuringTaskRolloverOnly`|If `true`, scale-down actions are deferred
until the next task rollover. This avoids disrupting in-progress ingestion.|No|
`false` |
+|`criticalLagThreshold`|Sets the aggregate-lag threshold for the high-lag and
critical-lag controls. At 75%, the autoscaler uses `highLagCostFactor` and
bypasses the scale-up task-count boundary. At 100%, it skips cost minimization
and jumps to `taskCountMax`. Must be greater than 0 when set. The threshold
uses aggregate lag units: records for Kafka and Rabbit, and milliseconds for
Kinesis.|No|`null` (disabled)|
+|`highLagCostFactor`|Cost factor applied when aggregate lag reaches 75% of
`criticalLagThreshold`. Higher values increase the lag cost during high lag.
Must be greater than or equal to 0. This property has no effect when
`criticalLagThreshold` is `null`.|No|`6.0`|
The following example shows a supervisor spec with `costBased` autoscaler:
@@ -234,7 +238,9 @@ The following example shows a supervisor spec with
`costBased` autoscaler:
"lagWeight": 0.4,
"idleWeight": 0.6,
"minScaleUpDelay": "PT10M",
- "minScaleDownDelay": "PT30M"
+ "minScaleDownDelay": "PT30M",
+ "criticalLagThreshold": 1000000,
+ "highLagCostFactor": 6.0
}
}
}
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 4a534b1a05d..b1955f8428f 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
@@ -213,6 +213,24 @@ public class CostBasedAutoScaler implements
SupervisorTaskAutoScaler
return config;
}
+ private boolean isHighLag(CostMetrics metrics)
+ {
+ final Long criticalLagThreshold = config.getCriticalLagThreshold();
+ return metrics != null && criticalLagThreshold != null
+ && metrics.getAggregateLag() >= criticalLagThreshold *
WeightedCostFunction.HIGH_LAG_THRESHOLD_FRACTION;
+ }
+
+ /**
+ * Whether the last collected metrics crossed {@link
CostBasedAutoScalerConfig#getCriticalLagThreshold()},
+ * meaning the argmin search should be skipped entirely in favor of jumping
to the maximum task count.
+ */
+ private boolean isCriticalLag(CostMetrics metrics)
+ {
+ final Long criticalLagThreshold = config.getCriticalLagThreshold();
+ return metrics != null && criticalLagThreshold != null
+ && metrics.getAggregateLag() >= criticalLagThreshold *
WeightedCostFunction.CRITICAL_LAG_THRESHOLD_FRACTION;
+ }
+
/**
* Returns the lowest-cost task count given {@code metrics}, or {@link
#CANNOT_COMPUTE} when
* metrics are unusable. Returning the current task count means the current
count is already
@@ -249,6 +267,24 @@ public class CostBasedAutoScaler implements
SupervisorTaskAutoScaler
return currentTaskCount;
}
+ final boolean highLag = isHighLag(metrics);
+ final boolean criticalLag = isCriticalLag(metrics);
+ if (criticalLag) {
+ log.info(
+ "Supervisor[%s] aggregateLag[%.0f] crossed [%.0f%%] of
criticalLagThreshold[%d]: skipping the argmin"
+ + " search and jumping straight to the maximum task count.",
+ supervisorId, metrics.getAggregateLag(),
WeightedCostFunction.CRITICAL_LAG_THRESHOLD_FRACTION * 100,
+ config.getCriticalLagThreshold()
+ );
+ } else if (highLag) {
+ log.info(
+ "Supervisor[%s] aggregateLag[%.0f] crossed [%.0f%%] of
criticalLagThreshold[%d]: widening scale-up"
+ + " candidates and maxing out the high-lag cost factor.",
+ supervisorId, metrics.getAggregateLag(),
WeightedCostFunction.HIGH_LAG_THRESHOLD_FRACTION * 100,
+ config.getCriticalLagThreshold()
+ );
+ }
+
// Start with the current task count as optimal
final CostResult currentCost = costFunction.computeCost(metrics,
currentTaskCount, config);
int optimalTaskCount = currentTaskCount;
@@ -277,7 +313,7 @@ public class CostBasedAutoScaler implements
SupervisorTaskAutoScaler
int startIndex = 0;
int endIndex = validTaskCounts.length - 1;
- if (config.isUseTaskCountBoundariesOnScaleUp()) {
+ if (config.isUseTaskCountBoundariesOnScaleUp() && !highLag) {
int currentTaskCountIndex = Arrays.binarySearch(validTaskCounts,
currentTaskCount);
endIndex = currentTaskCountIndex >= 0
? Math.min(currentTaskCountIndex +
BOUNDARY_LIMIT_IN_PARTITIONS_PER_TASK, endIndex)
@@ -291,13 +327,19 @@ public class CostBasedAutoScaler implements
SupervisorTaskAutoScaler
: startIndex;
}
+ // Critical lag skips the argmin search entirely: evaluate only the
maximum valid task count.
+ if (criticalLag) {
+ startIndex = validTaskCounts.length - 1;
+ endIndex = validTaskCounts.length - 1;
+ }
+
for (int i = startIndex; i <= endIndex; ++i) {
final int taskCount = validTaskCounts[i];
CostResult costResult = costFunction.computeCost(metrics, taskCount,
config);
double cost = costResult.totalCost();
costResults[i] = costResult;
- if (cost < optimalCost.totalCost()) {
+ if (criticalLag || cost < optimalCost.totalCost()) {
optimalTaskCount = taskCount;
optimalCost = costResult;
}
@@ -330,14 +372,16 @@ public class CostBasedAutoScaler implements
SupervisorTaskAutoScaler
emitter.emit(getMetricBuilder().setMetric(OPTIMAL_LAG_COST_METRIC,
optimalCost.lagCost()));
emitter.emit(getMetricBuilder().setMetric(OPTIMAL_IDLE_COST_METRIC,
optimalCost.idleCost()));
- final double costDropPercent
- = 100.0 * (currentCost.totalCost() - optimalCost.totalCost()) /
currentCost.totalCost();
- if (costDropPercent < config.getMinCostDropPercentForScaling()) {
- log.info(
- "Skipping scaling since cost drop percent[%.2f] is less than
required minCostDropPercentForScaling[%d]",
- costDropPercent, config.getMinCostDropPercentForScaling()
- );
- return currentTaskCount;
+ if (!criticalLag) {
+ final double costDropPercent
+ = 100.0 * (currentCost.totalCost() - optimalCost.totalCost()) /
currentCost.totalCost();
+ if (costDropPercent < config.getMinCostDropPercentForScaling()) {
+ log.info(
+ "Skipping scaling since cost drop percent[%.2f] is less than
required minCostDropPercentForScaling[%d]",
+ costDropPercent, config.getMinCostDropPercentForScaling()
+ );
+ return currentTaskCount;
+ }
}
}
@@ -476,11 +520,14 @@ public class CostBasedAutoScaler implements
SupervisorTaskAutoScaler
final LagStats lagStats = supervisor.computeLagStats();
final double avgPartitionLag;
+ final double aggregateLag;
if (lagStats == null) {
log.debug("Lag stats unavailable for supervisorId [%s], skipping
collection", supervisorId);
avgPartitionLag = -1;
+ aggregateLag = -1;
} else {
avgPartitionLag = lagStats.getAvgLag();
+ aggregateLag = lagStats.getTotalLag();
}
final int currentTaskCount = supervisor.getIoConfig().getTaskCount();
@@ -490,12 +537,13 @@ public class CostBasedAutoScaler implements
SupervisorTaskAutoScaler
final double movingAvgRate = extractMovingAverage(taskStats);
final double pollIdleRatio = extractPollIdleRatio(taskStats);
- if (!config.isUsePollIdleRatio() && movingAvgRate > 0) {
+ if (!config.isUsePollIdleRatio() && aggregateLag > 0 && movingAvgRate > 0)
{
processingRateSamples.add(movingAvgRate);
}
return new CostMetrics(
avgPartitionLag,
+ aggregateLag,
currentTaskCount,
partitionCount,
pollIdleRatio,
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 b93c354b6a5..92dda7e775e 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
@@ -46,8 +46,9 @@ public class CostBasedAutoScalerConfig implements
AutoScalerConfig
{
static final double DEFAULT_LAG_WEIGHT = 0.4;
static final double DEFAULT_IDLE_WEIGHT = 0.6;
- static final Duration DEFAULT_MIN_SCALE_UP_DELAY =
Duration.standardMinutes(10);
+ static final Duration DEFAULT_MIN_SCALE_UP_DELAY =
Duration.standardMinutes(15);
static final Duration DEFAULT_MIN_SCALE_DOWN_DELAY =
Duration.standardMinutes(30);
+ static final Duration DEFAULT_SCALE_ACTION_PERIOD =
Duration.standardMinutes(2);
private final boolean enableTaskAutoScaler;
private final int taskCountMax;
@@ -65,7 +66,9 @@ public class CostBasedAutoScalerConfig implements
AutoScalerConfig
private final Duration minScaleDownDelay;
private final boolean scaleDownDuringTaskRolloverOnly;
private final boolean usePollIdleRatio;
+ private final Long criticalLagThreshold;
private final int minCostDropPercentForScaling;
+ private final double highLagCostFactor;
/**
* Creates a new CostBasedAutoScalerConfig instance.
@@ -87,11 +90,13 @@ public class CostBasedAutoScalerConfig implements
AutoScalerConfig
@Nullable @JsonProperty("minScaleDownDelay") Duration minScaleDownDelay,
@Nullable @JsonProperty("scaleDownDuringTaskRolloverOnly") Boolean
scaleDownDuringTaskRolloverOnly,
@Nullable @JsonProperty("usePollIdleRatio") Boolean usePollIdleRatio,
- @Nullable @JsonProperty("minCostDropPercentForScaling") Integer
minCostDropPercentForScaling
+ @Nullable @JsonProperty("criticalLagThreshold") Long
criticalLagThreshold,
+ @Nullable @JsonProperty("minCostDropPercentForScaling") Integer
minCostDropPercentForScaling,
+ @Nullable @JsonProperty("highLagCostFactor") Double highLagCostFactor
)
{
this.enableTaskAutoScaler = Configs.valueOrDefault(enableTaskAutoScaler,
false);
- this.scaleActionPeriodMillis =
Configs.valueOrDefault(scaleActionPeriodMillis,
DEFAULT_MIN_SCALE_UP_DELAY.getMillis());
+ this.scaleActionPeriodMillis =
Configs.valueOrDefault(scaleActionPeriodMillis,
DEFAULT_SCALE_ACTION_PERIOD.getMillis());
this.lagWeight = Configs.valueOrDefault(lagWeight, DEFAULT_LAG_WEIGHT);
this.idleWeight = Configs.valueOrDefault(idleWeight, DEFAULT_IDLE_WEIGHT);
@@ -105,7 +110,25 @@ public class CostBasedAutoScalerConfig implements
AutoScalerConfig
this.minScaleDownDelay = Configs.valueOrDefault(minScaleDownDelay,
DEFAULT_MIN_SCALE_DOWN_DELAY);
this.scaleDownDuringTaskRolloverOnly =
Configs.valueOrDefault(scaleDownDuringTaskRolloverOnly, false);
this.usePollIdleRatio = Configs.valueOrDefault(usePollIdleRatio, true);
+ this.criticalLagThreshold = criticalLagThreshold;
+
+ Preconditions.checkArgument(
+ criticalLagThreshold == null || criticalLagThreshold > 0,
+ "criticalLagThreshold must be > 0"
+ );
this.minCostDropPercentForScaling =
Configs.valueOrDefault(minCostDropPercentForScaling, 0);
+ Preconditions.checkArgument(
+ this.minCostDropPercentForScaling >= 0 &&
this.minCostDropPercentForScaling <= 100,
+ "minCostDropPercentForScaling must be between 0 and 100"
+ );
+ this.highLagCostFactor = Configs.valueOrDefault(
+ highLagCostFactor,
+ WeightedCostFunction.DEFAULT_HIGH_LAG_COST_FACTOR
+ );
+ Preconditions.checkArgument(
+ this.highLagCostFactor >= 0,
+ "highLagCostFactor must be >= 0"
+ );
if (this.enableTaskAutoScaler) {
Preconditions.checkNotNull(taskCountMax, "taskCountMax is required when
enableTaskAutoScaler is true");
@@ -289,7 +312,25 @@ public class CostBasedAutoScalerConfig implements
AutoScalerConfig
}
/**
- * Minimum percentage drop from current cost that is required by the
auto-scaler
+ * Aggregate (sum-across-partitions) lag threshold driving a two-tier fast
path,
+ * relative to {@link CostMetrics#getAggregateLag()}:
+ * <ul>
+ * <li>At 75% of this value, the high-lag cost factor maxes out at 6.0
(instead of
+ * unamplified normal recovery), and the scale-up candidate search bypasses
+ * {@link #isUseTaskCountBoundariesOnScaleUp()}.</li>
+ * <li>At 100% of this value, cost minimization is skipped entirely and
the task count jumps
+ * straight to the maximum.</li>
+ * </ul>
+ * {@code null} disables the feature.
+ */
+ @JsonProperty
+ @Nullable
+ public Long getCriticalLagThreshold()
+ {
+ return criticalLagThreshold;
+ }
+
+ /* Minimum percentage drop from current cost that is required by the
auto-scaler
* to choose a new task count.
*/
@JsonProperty
@@ -298,6 +339,12 @@ public class CostBasedAutoScalerConfig implements
AutoScalerConfig
return minCostDropPercentForScaling;
}
+ @JsonProperty
+ public double getHighLagCostFactor()
+ {
+ return highLagCostFactor;
+ }
+
@Override
public SupervisorTaskAutoScaler createAutoScaler(Supervisor supervisor,
SupervisorSpec spec, ServiceEmitter emitter)
{
@@ -330,8 +377,10 @@ public class CostBasedAutoScalerConfig implements
AutoScalerConfig
&& scaleDownDuringTaskRolloverOnly ==
that.scaleDownDuringTaskRolloverOnly
&& usePollIdleRatio == that.usePollIdleRatio
&& minCostDropPercentForScaling == that.minCostDropPercentForScaling
+ && Double.compare(that.highLagCostFactor, highLagCostFactor) == 0
&& Objects.equals(taskCountStart, that.taskCountStart)
- && Objects.equals(stopTaskCountRatio, that.stopTaskCountRatio);
+ && Objects.equals(stopTaskCountRatio, that.stopTaskCountRatio)
+ && Objects.equals(criticalLagThreshold, that.criticalLagThreshold);
}
@Override
@@ -353,7 +402,9 @@ public class CostBasedAutoScalerConfig implements
AutoScalerConfig
minScaleDownDelay,
scaleDownDuringTaskRolloverOnly,
usePollIdleRatio,
- minCostDropPercentForScaling
+ criticalLagThreshold,
+ minCostDropPercentForScaling,
+ highLagCostFactor
);
}
@@ -376,7 +427,9 @@ public class CostBasedAutoScalerConfig implements
AutoScalerConfig
", minScaleDownDelay=" + minScaleDownDelay +
", scaleDownDuringTaskRolloverOnly=" +
scaleDownDuringTaskRolloverOnly +
", usePollIdleRatio=" + usePollIdleRatio +
+ ", criticalLagThreshold=" + criticalLagThreshold +
", minCostDropPercentForScaling=" + minCostDropPercentForScaling +
+ ", highLagCostFactor=" + highLagCostFactor +
'}';
}
@@ -401,7 +454,9 @@ public class CostBasedAutoScalerConfig implements
AutoScalerConfig
private Duration minScaleDownDelay;
private Boolean scaleDownDuringTaskRolloverOnly;
private Boolean usePollIdleRatio;
+ private Long criticalLagThreshold;
private Integer minCostDropPercentForScaling;
+ private Double highLagCostFactor;
private Builder()
{
@@ -485,6 +540,12 @@ public class CostBasedAutoScalerConfig implements
AutoScalerConfig
return this;
}
+ public Builder criticalLagThreshold(Long criticalLagThreshold)
+ {
+ this.criticalLagThreshold = criticalLagThreshold;
+ return this;
+ }
+
public Builder useTaskCountBoundariesOnScaleUp(boolean
useTaskCountBoundariesOnScaleUp)
{
this.useTaskCountBoundariesOnScaleUp = useTaskCountBoundariesOnScaleUp;
@@ -503,6 +564,12 @@ public class CostBasedAutoScalerConfig implements
AutoScalerConfig
return this;
}
+ public Builder highLagCostFactor(double highLagCostFactor)
+ {
+ this.highLagCostFactor = highLagCostFactor;
+ return this;
+ }
+
public CostBasedAutoScalerConfig build()
{
return new CostBasedAutoScalerConfig(
@@ -521,7 +588,9 @@ public class CostBasedAutoScalerConfig implements
AutoScalerConfig
minScaleDownDelay,
scaleDownDuringTaskRolloverOnly,
usePollIdleRatio,
- minCostDropPercentForScaling
+ criticalLagThreshold,
+ minCostDropPercentForScaling,
+ highLagCostFactor
);
}
}
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 33bf7eff33d..f3e89849766 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
@@ -41,6 +41,7 @@ public class CostMetrics
public CostMetrics(
double avgPartitionLag,
+ double aggregateLag,
int currentTaskCount,
int partitionCount,
double pollIdleRatio,
@@ -55,7 +56,7 @@ public class CostMetrics
this.pollIdleRatio = pollIdleRatio;
this.taskDurationSeconds = taskDurationSeconds;
this.avgProcessingRate = avgProcessingRate;
- this.aggregateLag = avgPartitionLag * partitionCount;
+ this.aggregateLag = aggregateLag;
this.maxObservedRate = maxObservedRate;
}
@@ -89,7 +90,6 @@ public class CostMetrics
/**
* Returns the aggregated lag across all partitions.
- * Pre-computed as avgPartitionLag * partitionCount.
*/
public double getAggregateLag()
{
@@ -142,6 +142,7 @@ public class CostMetrics
}
CostMetrics that = (CostMetrics) o;
return Double.compare(that.avgPartitionLag, avgPartitionLag) == 0
+ && Double.compare(that.aggregateLag, aggregateLag) == 0
&& currentTaskCount == that.currentTaskCount
&& partitionCount == that.partitionCount
&& Double.compare(that.pollIdleRatio, pollIdleRatio) == 0
@@ -155,6 +156,7 @@ public class CostMetrics
{
return Objects.hash(
avgPartitionLag,
+ aggregateLag,
currentTaskCount,
partitionCount,
pollIdleRatio,
@@ -169,6 +171,7 @@ public class CostMetrics
{
return "CostMetrics{" +
"avgPartitionLag=" + avgPartitionLag +
+ ", aggregateLag=" + aggregateLag +
", currentTaskCount=" + currentTaskCount +
", partitionCount=" + partitionCount +
", pollIdleRatio=" + pollIdleRatio +
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 beaf0a5b9d5..b0a5a21c06c 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
@@ -37,10 +37,30 @@ public class WeightedCostFunction
private static final Logger log = new Logger(WeightedCostFunction.class);
/**
- * Multiplier for a lag amplification factor; it was carefully chosen
- * during extensive testing as the most balanced multiplier for high-lag
recovery.
+ * Normal-path lag amplification multiplier. Critical-lag tiers provide the
+ * urgency amplification, so normal lag uses unamplified recovery time.
*/
- static final double LAG_AMPLIFICATION_MULTIPLIER = 0.3;
+ static final double LAG_AMPLIFICATION_MULTIPLIER = 0.0;
+
+ /**
+ * Default cost factor used once aggregate lag crosses {@link
#HIGH_LAG_THRESHOLD_FRACTION} of
+ * {@link CostBasedAutoScalerConfig#getCriticalLagThreshold()}.
+ */
+ static final double DEFAULT_HIGH_LAG_COST_FACTOR = 6.0;
+
+ /**
+ * Fraction of {@link CostBasedAutoScalerConfig#getCriticalLagThreshold()}
at which the high-lag
+ * fast path engages: the cost factor maxes out at {@link
#DEFAULT_HIGH_LAG_COST_FACTOR}
+ * and the scale-up candidate boundary is bypassed.
+ */
+ static final double HIGH_LAG_THRESHOLD_FRACTION = 0.75;
+
+ /**
+ * Fraction of {@link CostBasedAutoScalerConfig#getCriticalLagThreshold()}
at which the critical-lag
+ * fast path engages: the cost-minimization search is skipped entirely and
the task count jumps
+ * straight to the maximum.
+ */
+ static final double CRITICAL_LAG_THRESHOLD_FRACTION = 1.0;
/**
* Exponent (< 1) for sublinear busy redistribution in the idle projection:
@@ -106,14 +126,20 @@ public class WeightedCostFunction
}
// Lag recovery time is decreasing by adding tasks and increasing by
ejecting tasks.
- // In case of increasing lag, we apply an amplification factor to reflect
the urgency of addressing lag.
- // Caution: we rely only on the metrics, the real issues may be absolutely
different, up to hardware failure.
+ // High lag uses extra cost; normal lag uses raw recovery time so capacity
cost remains meaningful.
+ // Once aggregate lag crosses HIGH_LAG_THRESHOLD_FRACTION of
criticalLagThreshold, the cost factor is
+ // maxed out at the configured highLagCostFactor (vs unamplified normal
recovery).
final double lagRecoveryTime;
if (metrics.getAggregateLag() <= 0) {
lagRecoveryTime = 0;
} else {
final double lagPerPartition = metrics.getAggregateLag() /
metrics.getPartitionCount();
- final double amplification = Math.max(1.0, 1.0 +
LAG_AMPLIFICATION_MULTIPLIER * Math.log(lagPerPartition));
+ final Long criticalLagThreshold = config.getCriticalLagThreshold();
+ final boolean highLag = criticalLagThreshold != null
+ && metrics.getAggregateLag() >= criticalLagThreshold *
HIGH_LAG_THRESHOLD_FRACTION;
+
+ final double costFactor = highLag ? config.getHighLagCostFactor() :
LAG_AMPLIFICATION_MULTIPLIER;
+ final double amplification = Math.max(1.0, 1.0 + costFactor *
Math.log(lagPerPartition));
final double adjustedProcessingRate = Math.max(avgProcessingRate,
MIN_PROCESSING_RATE);
lagRecoveryTime = metrics.getAggregateLag() * amplification /
(proposedTaskCount * adjustedProcessingRate);
}
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 295dd1efe88..3550094d994 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
@@ -29,6 +29,8 @@ import static
org.apache.druid.indexing.seekablestream.supervisor.autoscaler.Cos
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_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.CostBasedAutoScalerConfig.DEFAULT_SCALE_ACTION_PERIOD;
+import static
org.apache.druid.indexing.seekablestream.supervisor.autoscaler.WeightedCostFunction.DEFAULT_HIGH_LAG_COST_FACTOR;
import static
org.apache.druid.indexing.seekablestream.supervisor.autoscaler.WeightedCostFunction.OPTIMAL_TASK_IDLE_RATIO;
@SuppressWarnings("TextBlockMigration")
@@ -55,7 +57,9 @@ public class CostBasedAutoScalerConfigTest
+ " \"minScaleDownDelay\": \"PT10M\",\n"
+ " \"scaleDownDuringTaskRolloverOnly\": true,\n"
+ " \"usePollIdleRatio\": false,\n"
- + " \"minCostDropPercentForScaling\": 10\n"
+ + " \"criticalLagThreshold\": 500000,\n"
+ + " \"minCostDropPercentForScaling\": 10,\n"
+ + " \"highLagCostFactor\": 8.0\n"
+ "}";
final CostBasedAutoScalerConfig config = mapper.readValue(json,
CostBasedAutoScalerConfig.class);
@@ -75,7 +79,9 @@ public class CostBasedAutoScalerConfigTest
Assert.assertFalse(config.isUsePollIdleRatio());
Assert.assertFalse(config.isUseTaskCountBoundariesOnScaleUp());
Assert.assertTrue(config.isUseTaskCountBoundariesOnScaleDown());
+ Assert.assertEquals(Long.valueOf(500000),
config.getCriticalLagThreshold());
Assert.assertEquals(10, config.getMinCostDropPercentForScaling());
+ Assert.assertEquals(8.0, config.getHighLagCostFactor(), 0.001);
// Test serialization back to JSON
final String serialized = mapper.writeValueAsString(config);
@@ -101,7 +107,7 @@ public class CostBasedAutoScalerConfigTest
Assert.assertEquals(2, config.getTaskCountMin());
// Check defaults
- Assert.assertEquals(DEFAULT_MIN_SCALE_UP_DELAY.getMillis(),
config.getScaleActionPeriodMillis());
+ Assert.assertEquals(DEFAULT_SCALE_ACTION_PERIOD.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);
@@ -114,7 +120,9 @@ public class CostBasedAutoScalerConfigTest
Assert.assertTrue(config.isUseTaskCountBoundariesOnScaleDown());
Assert.assertNull(config.getTaskCountStart());
Assert.assertNull(config.getStopTaskCountRatio());
+ Assert.assertNull(config.getCriticalLagThreshold());
Assert.assertEquals(0, config.getMinCostDropPercentForScaling());
+ Assert.assertEquals(DEFAULT_HIGH_LAG_COST_FACTOR,
config.getHighLagCostFactor(), 0.001);
}
@Test
@@ -224,6 +232,8 @@ public class CostBasedAutoScalerConfigTest
.minScaleDownDelay(Duration.standardMinutes(10))
.scaleDownDuringTaskRolloverOnly(true)
.usePollIdleRatio(false)
+
.criticalLagThreshold(500000L)
+
.highLagCostFactor(8.0)
.build();
Assert.assertTrue(config.getEnableTaskAutoScaler());
@@ -241,6 +251,52 @@ public class CostBasedAutoScalerConfigTest
Assert.assertEquals(Duration.standardMinutes(10),
config.getMinScaleDownDelay());
Assert.assertTrue(config.isScaleDownOnTaskRolloverOnly());
Assert.assertFalse(config.isUsePollIdleRatio());
+ Assert.assertEquals(Long.valueOf(500000),
config.getCriticalLagThreshold());
+ Assert.assertEquals(8.0, config.getHighLagCostFactor(), 0.001);
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testValidation_NegativeCriticalLagAmplificationMultiplier()
+ {
+ CostBasedAutoScalerConfig.builder()
+ .taskCountMax(100)
+ .taskCountMin(5)
+ .highLagCostFactor(-1.0)
+ .enableTaskAutoScaler(true)
+ .build();
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testValidation_ZeroCriticalLagThreshold()
+ {
+ CostBasedAutoScalerConfig.builder()
+ .taskCountMax(100)
+ .taskCountMin(5)
+ .criticalLagThreshold(0L)
+ .enableTaskAutoScaler(true)
+ .build();
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testValidation_NegativeMinCostDropPercentForScaling()
+ {
+ CostBasedAutoScalerConfig.builder()
+ .taskCountMax(100)
+ .taskCountMin(5)
+ .minCostDropPercentForScaling(-1)
+ .enableTaskAutoScaler(true)
+ .build();
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testValidation_MinCostDropPercentForScalingAbove100()
+ {
+ CostBasedAutoScalerConfig.builder()
+ .taskCountMax(100)
+ .taskCountMin(5)
+ .minCostDropPercentForScaling(101)
+ .enableTaskAutoScaler(true)
+ .build();
}
@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 fa86fbf1d3c..3430842517d 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
@@ -367,6 +367,7 @@ public class CostBasedAutoScalerMockTest
{
CostMetrics metrics = new CostMetrics(
avgLag,
+ avgLag * PARTITION_COUNT,
taskCount,
PARTITION_COUNT,
pollIdleRatio,
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 435427fa37b..4441273d9a0 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
@@ -240,6 +240,135 @@ public class CostBasedAutoScalerTest
);
}
+ @Test
+ public void testHighLagThresholdBypassesScaleUpBoundary()
+ {
+ // aggregateLag = 100_000 * 100 = 10,000,000. With threshold=12,000,000:
tier1=9,000,000 (crossed),
+ // critical lag=11,400,000 (not crossed), so this exercises high lag
(boundary bypass) without
+ // triggering the critical-lag jump-to-max.
+ final CostBasedAutoScalerConfig boundedScaleUpConfig =
CostBasedAutoScalerConfig
+ .builder()
+ .taskCountMax(100)
+ .taskCountMin(1)
+ .enableTaskAutoScaler(true)
+ .lagWeight(1.0)
+ .idleWeight(0.0)
+ .useTaskCountBoundariesOnScaleUp(true)
+ .criticalLagThreshold(12_000_000L)
+ .build();
+ final CostBasedAutoScaler scaler = createAutoScaler(boundedScaleUpConfig);
+
+ Assert.assertEquals(
+ "High lag should bypass the scale-up boundary and jump straight to the
argmin",
+ 100,
+ scaler.computeOptimalTaskCount(createMetrics(100_000.0, 10, 100, 0.25))
+ );
+
+ // Below the threshold, the boundary still applies as usual.
+ Assert.assertEquals(
+ "Below criticalLagThreshold, the scale-up boundary still limits
candidates",
+ 13,
+ scaler.computeOptimalTaskCount(createMetrics(10.0, 10, 100, 0.25))
+ );
+ }
+
+ @Test
+ public void testHighLagThresholdUsesExactAggregateLag()
+ {
+ final CostBasedAutoScalerConfig boundedScaleUpConfig =
CostBasedAutoScalerConfig
+ .builder()
+ .taskCountMax(1_000)
+ .taskCountMin(1)
+ .enableTaskAutoScaler(true)
+ .lagWeight(1.0)
+ .idleWeight(0.0)
+ .useTaskCountBoundariesOnScaleUp(true)
+ .criticalLagThreshold(1_000L)
+ .build();
+ final CostBasedAutoScaler scaler = createAutoScaler(boundedScaleUpConfig);
+
+ Assert.assertEquals(
+ "Exact aggregate lag should engage high lag even when integer average
lag is zero",
+ 1_000,
+ scaler.computeOptimalTaskCount(createMetrics(0.0, 999.0, 10, 1_000,
0.25))
+ );
+ }
+
+ @Test
+ public void testCriticalLagJumpsStraightToMaxTaskCount()
+ {
+ // aggregateLag = 100_000 * 500 = 50,000,000. With threshold=10,000,000:
tier2=10,000,000 is
+ // comfortably crossed, so the argmin search is skipped entirely in favor
of the maximum task count.
+ final CostBasedAutoScalerConfig config = CostBasedAutoScalerConfig
+ .builder()
+ .taskCountMax(500)
+ .taskCountMin(1)
+ .enableTaskAutoScaler(true)
+ .lagWeight(0.1)
+ .idleWeight(0.9)
+ .criticalLagThreshold(10_000_000L)
+ .build();
+ final CostBasedAutoScaler scaler = createAutoScaler(config);
+
+ // Idle-heavy weights would normally argue for scaling down, but critical
lag overrides that entirely.
+ Assert.assertEquals(
+ "Critical lag should jump straight to the maximum task count
regardless of idle-favoring weights",
+ 500,
+ scaler.computeOptimalTaskCount(createMetrics(100_000.0, 10, 500, 0.9))
+ );
+ }
+
+ @Test
+ public void testCriticalLagJumpsToMaxEvenWhenMaxCostsMore()
+ {
+ // lagWeight=0 means the max candidate's cost is driven entirely by idle
cost, which is higher
+ // at 500 tasks than at the current 10 tasks. Critical lag must still jump
to the maximum
+ // instead of leaving the current (cheaper-looking) task count in place.
+ final CostBasedAutoScalerConfig config = CostBasedAutoScalerConfig
+ .builder()
+ .taskCountMax(500)
+ .taskCountMin(1)
+ .enableTaskAutoScaler(true)
+ .lagWeight(0.0)
+ .idleWeight(1.0)
+ .criticalLagThreshold(10_000_000L)
+ .build();
+ final CostBasedAutoScaler scaler = createAutoScaler(config);
+
+ Assert.assertEquals(
+ "Critical lag should jump to the maximum task count even if it costs
more than the current count",
+ 500,
+ scaler.computeOptimalTaskCount(createMetrics(100_000.0, 10, 500, 0.9))
+ );
+ }
+
+ @Test
+ public void testCriticalLagRequiresFullThreshold()
+ {
+ final CostBasedAutoScalerConfig config = CostBasedAutoScalerConfig
+ .builder()
+ .taskCountMax(500)
+ .taskCountMin(1)
+ .enableTaskAutoScaler(true)
+ .lagWeight(0.0)
+ .idleWeight(1.0)
+ .useTaskCountBoundariesOnScaleDown(false)
+ .criticalLagThreshold(10_000_000L)
+ .build();
+ final CostBasedAutoScaler scaler = createAutoScaler(config);
+
+ Assert.assertNotEquals(
+ "Tier 2 should not trigger below the full critical lag threshold",
+ 500,
+ scaler.computeOptimalTaskCount(createMetrics(20_000.0, 9_999_999.0,
10, 500, 0.9))
+ );
+ Assert.assertEquals(
+ "Tier 2 should trigger at the full critical lag threshold",
+ 500,
+ scaler.computeOptimalTaskCount(createMetrics(20_000.0, 10_000_000.0,
10, 500, 0.9))
+ );
+ }
+
@Test
public void testExtractPollIdleRatio()
{
@@ -553,6 +682,35 @@ public class CostBasedAutoScalerTest
);
}
+ @Test
+ public void testCollectMetricsPreservesExactAggregateLag()
+ {
+ final SupervisorSpec spec = Mockito.mock(SupervisorSpec.class);
+ final SeekableStreamSupervisor supervisor =
Mockito.mock(SeekableStreamSupervisor.class);
+ final ServiceEmitter emitter = Mockito.mock(ServiceEmitter.class);
+ final 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(ioConfig.getTaskCount()).thenReturn(10);
+ when(supervisor.getPartitionCount()).thenReturn(1_000);
+ when(supervisor.computeLagStats()).thenReturn(new LagStats(999, 999, 0));
+ when(supervisor.getStats()).thenReturn(Collections.emptyMap());
+
+ final CostBasedAutoScalerConfig config =
CostBasedAutoScalerConfig.builder()
+
.taskCountMax(1_000)
+
.taskCountMin(1)
+
.enableTaskAutoScaler(true)
+
.build();
+ final CostBasedAutoScaler scaler = new CostBasedAutoScaler(supervisor,
config, spec, emitter);
+
+ Assert.assertEquals(999.0, scaler.collectMetrics().getAggregateLag(), 0.0);
+ }
+
@Test
public void
testCollectMetricsTracksMaxProcessingRateOnlyWhenPollIdleRatioDisabled()
{
@@ -568,7 +726,7 @@ public class CostBasedAutoScalerTest
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));
+ when(supervisor.computeLagStats()).thenReturn(new LagStats(1, 1, 0));
// usePollIdleRatio defaults to true, which disables rate-watermark
tracking entirely.
CostBasedAutoScalerConfig defaultConfig =
CostBasedAutoScalerConfig.builder()
@@ -599,9 +757,19 @@ public class CostBasedAutoScalerTest
CostBasedAutoScaler autoScalerWithoutPollIdleRatio =
new CostBasedAutoScaler(supervisor, configWithoutPollIdleRatio, spec,
emitter);
+ when(supervisor.computeLagStats()).thenReturn(
+ new LagStats(0, 0, 0),
+ new LagStats(1, 1, 0),
+ new LagStats(1, 1, 0)
+ );
+
+ Assert.assertNull(
+ "A rate sample without lag must not establish the watermark",
+ autoScalerWithoutPollIdleRatio.collectMetrics().getMaxObservedRate()
+ );
Assert.assertEquals(
- "First sample becomes the watermark",
- 500.0,
+ "First positive-lag sample becomes the watermark",
+ 9000.0,
autoScalerWithoutPollIdleRatio.collectMetrics().getMaxObservedRate(),
0.0001
);
@@ -635,6 +803,27 @@ public class CostBasedAutoScalerTest
{
return new CostMetrics(
avgPartitionLag,
+ avgPartitionLag * partitionCount,
+ currentTaskCount,
+ partitionCount,
+ pollIdleRatio,
+ 3600,
+ 1000.0,
+ 0.
+ );
+ }
+
+ private CostMetrics createMetrics(
+ double avgPartitionLag,
+ double aggregateLag,
+ int currentTaskCount,
+ int partitionCount,
+ double pollIdleRatio
+ )
+ {
+ return new CostMetrics(
+ avgPartitionLag,
+ aggregateLag,
currentTaskCount,
partitionCount,
pollIdleRatio,
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 6802692ade4..3399fb80316 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
@@ -329,7 +329,7 @@ public class WeightedCostFunctionTest
}
@Test
- public void testLagAmplificationAppliedUnconditionally()
+ public void testNormalLagCostUsesUnamplifiedRecoveryTime()
{
CostBasedAutoScalerConfig lagOnly = CostBasedAutoScalerConfig.builder()
.taskCountMax(100)
@@ -344,23 +344,83 @@ public class WeightedCostFunctionTest
int partitionCount = 10;
double pollIdleRatio = 0.1;
- // lagPerPartition = 150 * 10 / 10 = 150, amplification = 1 + 0.2 * ln(150)
+ // Normal lag uses raw recovery time; high lag is tested separately below.
CostMetrics metrics = createMetrics(150.0, currentTaskCount,
partitionCount, pollIdleRatio);
double costWithAmp = costFunction.computeCost(metrics, proposedTaskCount,
lagOnly).totalCost();
double aggregateLag = 150.0 * partitionCount;
- double lagPerPartition = aggregateLag / partitionCount;
- double amplification = 1.0 +
WeightedCostFunction.LAG_AMPLIFICATION_MULTIPLIER * Math.log(lagPerPartition);
- double expected = aggregateLag * amplification / (proposedTaskCount *
WeightedCostFunction.MIN_PROCESSING_RATE);
+ double expected = aggregateLag / (proposedTaskCount *
WeightedCostFunction.MIN_PROCESSING_RATE);
+
+ Assert.assertEquals("Normal lag cost should use raw recovery time",
expected, costWithAmp, 0.0001);
+ }
+
+ @Test
+ public void testHighLagThresholdMaxesOutCostFactor()
+ {
+ int currentTaskCount = 10;
+ int proposedTaskCount = 10;
+ int partitionCount = 10;
+ double avgPartitionLag = 150.0;
+ double aggregateLag = avgPartitionLag * partitionCount;
+
+ CostMetrics metrics = createMetrics(avgPartitionLag, currentTaskCount,
partitionCount, 0.1);
+
+ // aggregateLag sits at exactly tier1Fraction (75%) of this threshold.
+ long tier1Threshold = (long) (aggregateLag /
WeightedCostFunction.HIGH_LAG_THRESHOLD_FRACTION);
+
+ CostBasedAutoScalerConfig noThreshold = CostBasedAutoScalerConfig.builder()
+
.taskCountMax(100)
+
.taskCountMin(1)
+
.enableTaskAutoScaler(true)
+
.lagWeight(1.0)
+
.idleWeight(0.0)
+ .build();
+ CostBasedAutoScalerConfig belowTier1 = CostBasedAutoScalerConfig.builder()
+
.taskCountMax(100)
+
.taskCountMin(1)
+
.enableTaskAutoScaler(true)
+
.lagWeight(1.0)
+
.idleWeight(0.0)
+
.criticalLagThreshold(tier1Threshold + 100)
+ .build();
+ CostBasedAutoScalerConfig atTier1 = CostBasedAutoScalerConfig.builder()
+
.taskCountMax(100)
+
.taskCountMin(1)
+
.enableTaskAutoScaler(true)
+
.lagWeight(1.0)
+
.idleWeight(0.0)
+
.criticalLagThreshold(tier1Threshold)
+ .build();
- Assert.assertEquals("Lag amplification should increase lag recovery time",
expected, costWithAmp, 0.0001);
+ double costBelowTier1 = costFunction.computeCost(metrics,
proposedTaskCount, belowTier1).totalCost();
+ Assert.assertEquals(
+ "Below tier1, amplification uses the default multiplier",
+ costFunction.computeCost(metrics, proposedTaskCount,
noThreshold).totalCost(),
+ costBelowTier1,
+ 0.0001
+ );
+
+ double lagPerPartition = aggregateLag / partitionCount;
+ double highLagCostFactor =
+ 1.0 + WeightedCostFunction.DEFAULT_HIGH_LAG_COST_FACTOR *
Math.log(lagPerPartition);
+ double costAtTier1 = costFunction.computeCost(metrics, proposedTaskCount,
atTier1).totalCost();
+ Assert.assertEquals(
+ "At/above the high-lag threshold, the cost factor maxes out at
DEFAULT_HIGH_LAG_COST_FACTOR",
+ aggregateLag * highLagCostFactor / (proposedTaskCount *
WeightedCostFunction.MIN_PROCESSING_RATE),
+ costAtTier1,
+ 0.0001
+ );
+ Assert.assertTrue(
+ "High-lag cost should exceed the default-multiplier cost for the same
lag",
+ costAtTier1 > costBelowTier1
+ );
}
@Test
- public void testAmplificationGrowsWithLag()
+ public void testNormalLagCostScalesLinearlyWithLag()
{
- // Verify that higher lag produces proportionally higher cost due to log
amplification
+ // Without normal-path amplification, cost grows linearly with lag.
CostBasedAutoScalerConfig lagOnly = CostBasedAutoScalerConfig.builder()
.taskCountMax(100)
.taskCountMin(1)
@@ -382,12 +442,14 @@ public class WeightedCostFunctionTest
Assert.assertTrue("Higher lag should produce higher cost", highCost >
lowCost);
- // The ratio of costs should be more than the ratio of raw lags (due to
amplification)
+ // The ratio of costs matches the ratio of raw lags.
double lagRatio = 10_000.0 / 100.0;
double costRatio = highCost / lowCost;
- Assert.assertTrue(
- "Amplification should make cost grow faster than linear with lag",
- costRatio > lagRatio
+ Assert.assertEquals(
+ "Normal lag cost should grow linearly with lag",
+ lagRatio,
+ costRatio,
+ 0.0001
);
}
@@ -541,6 +603,7 @@ public class WeightedCostFunctionTest
{
return new CostMetrics(
avgPartitionLag,
+ avgPartitionLag * partitionCount,
currentTaskCount,
partitionCount,
pollIdleRatio,
@@ -556,7 +619,7 @@ public class WeightedCostFunctionTest
double pollIdleRatio
)
{
- return new CostMetrics(0.0, 10, 100, pollIdleRatio, 3600,
avgProcessingRate, maxObservedRate);
+ return new CostMetrics(0.0, 0.0, 10, 100, pollIdleRatio, 3600,
avgProcessingRate, maxObservedRate);
}
private CostMetrics createMetricsWithRate(
@@ -569,6 +632,7 @@ public class WeightedCostFunctionTest
{
return new CostMetrics(
avgPartitionLag,
+ avgPartitionLag * partitionCount,
currentTaskCount,
partitionCount,
pollIdleRatio,
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
