Skip to content

Commit 61b39cf

Browse files
author
huyuanfeng
committed
[FLINK-36836][Autoscaler] Remove the default value of max min, calculated by utilization.target.boundary 
1 parent 65c53e7 commit 61b39cf

File tree

5 files changed

+62
-23
lines changed

5 files changed

+62
-23
lines changed

‎docs/layouts/shortcodes/generated/auto_scaler_configuration.html‎

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -202,13 +202,13 @@
202202
</tr>
203203
<tr>
204204
<td><h5>job.autoscaler.utilization.max</h5></td>
205-
<td style="word-wrap: break-word;">1.0</td>
205+
<td style="word-wrap: break-word;">(none)</td>
206206
<td>Double</td>
207207
<td>Max vertex utilization</td>
208208
</tr>
209209
<tr>
210210
<td><h5>job.autoscaler.utilization.min</h5></td>
211-
<td style="word-wrap: break-word;">0.4</td>
211+
<td style="word-wrap: break-word;">(none)</td>
212212
<td>Double</td>
213213
<td>Min vertex utilization</td>
214214
</tr>
@@ -218,6 +218,12 @@
218218
<td>Double</td>
219219
<td>Target vertex utilization</td>
220220
</tr>
221+
<tr>
222+
<td><h5>job.autoscaler.utilization.target.boundary</h5></td>
223+
<td style="word-wrap: break-word;">0.3</td>
224+
<td>Double</td>
225+
<td>Target vertex utilization boundary. Scaling won't be performed if the processing capacity is within [target_rate / (target_utilization - boundary), (target_rate / (target_utilization + boundary)]</td>
226+
</tr>
221227
<tr>
222228
<td><h5>job.autoscaler.vertex.exclude.ids</h5></td>
223229
<td style="word-wrap: break-word;"></td>

‎flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricEvaluator.java‎

Lines changed: 8 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -44,10 +44,10 @@
4444
import java.util.SortedMap;
4545

4646
import static org.apache.flink.autoscaler.config.AutoScalerOptions.BACKLOG_PROCESSING_LAG_THRESHOLD;
47-
import static org.apache.flink.autoscaler.config.AutoScalerOptions.TARGET_UTILIZATION_BOUNDARY;
4847
import static org.apache.flink.autoscaler.config.AutoScalerOptions.UTILIZATION_MAX;
4948
import static org.apache.flink.autoscaler.config.AutoScalerOptions.UTILIZATION_MIN;
5049
import static org.apache.flink.autoscaler.config.AutoScalerOptions.UTILIZATION_TARGET;
50+
import static org.apache.flink.autoscaler.config.AutoScalerOptions.UTILIZATION_TARGET_BOUNDARY;
5151
import static org.apache.flink.autoscaler.metrics.ScalingMetric.CATCH_UP_DATA_RATE;
5252
import static org.apache.flink.autoscaler.metrics.ScalingMetric.GC_PRESSURE;
5353
import static org.apache.flink.autoscaler.metrics.ScalingMetric.HEAP_MAX_USAGE_RATIO;
@@ -287,6 +287,7 @@ protected static void computeProcessingRateThresholds(
287287
Duration restartTime) {
288288

289289
double targetUtilization = conf.get(UTILIZATION_TARGET);
290+
double utilizationBoundary = conf.get(UTILIZATION_TARGET_BOUNDARY);
290291

291292
double upperUtilization;
292293
double lowerUtilization;
@@ -297,16 +298,12 @@ protected static void computeProcessingRateThresholds(
297298
upperUtilization = 1.0;
298299
lowerUtilization = 0.0;
299300
} else {
300-
if (conf.getOptional(UTILIZATION_MAX).isPresent()
301-
|| conf.getOptional(UTILIZATION_MIN).isPresent()
302-
|| conf.getOptional(TARGET_UTILIZATION_BOUNDARY).isEmpty()) {
303-
upperUtilization = conf.get(UTILIZATION_MAX);
304-
lowerUtilization = conf.get(UTILIZATION_MIN);
305-
} else {
306-
Double boundary = conf.get(TARGET_UTILIZATION_BOUNDARY);
307-
upperUtilization = targetUtilization + boundary;
308-
lowerUtilization = targetUtilization - boundary;
309-
}
301+
upperUtilization =
302+
conf.getOptional(UTILIZATION_MAX)
303+
.orElse(targetUtilization + utilizationBoundary);
304+
lowerUtilization =
305+
conf.getOptional(UTILIZATION_MIN)
306+
.orElse(targetUtilization - utilizationBoundary);
310307
}
311308

312309
double scaleUpThreshold =

‎flink-autoscaler/src/main/java/org/apache/flink/autoscaler/config/AutoScalerOptions.java‎

Lines changed: 7 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -99,26 +99,28 @@ private static ConfigOptions.OptionBuilder autoScalerConfig(String key) {
9999
oldOperatorConfigKey("target.utilization"))
100100
.withDescription("Target vertex utilization");
101101

102-
@Deprecated
103-
public static final ConfigOption<Double> TARGET_UTILIZATION_BOUNDARY =
104-
autoScalerConfig("target.utilization.boundary")
102+
public static final ConfigOption<Double> UTILIZATION_TARGET_BOUNDARY =
103+
autoScalerConfig("utilization.target.boundary")
105104
.doubleType()
106105
.defaultValue(0.3)
107106
.withFallbackKeys(oldOperatorConfigKey("target.utilization.boundary"))
107+
.withFallbackKeys(
108+
oldOperatorConfigKey("target.utilization.boundary"),
109+
oldOperatorConfigKey("utilization.target.boundary"))
108110
.withDescription(
109111
"Target vertex utilization boundary. Scaling won't be performed if the processing capacity is within [target_rate / (target_utilization - boundary), (target_rate / (target_utilization + boundary)]");
110112

111113
public static final ConfigOption<Double> UTILIZATION_MAX =
112114
autoScalerConfig("utilization.max")
113115
.doubleType()
114-
.defaultValue(1.)
116+
.noDefaultValue()
115117
.withFallbackKeys(oldOperatorConfigKey("utilization.max"))
116118
.withDescription("Max vertex utilization");
117119

118120
public static final ConfigOption<Double> UTILIZATION_MIN =
119121
autoScalerConfig("utilization.min")
120122
.doubleType()
121-
.defaultValue(0.4)
123+
.noDefaultValue()
122124
.withFallbackKeys(oldOperatorConfigKey("utilization.min"))
123125
.withDescription("Min vertex utilization");
124126

‎flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingExecutorTest.java‎

Lines changed: 38 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -265,7 +265,7 @@ public void testUtilizationBoundariesAndUtilizationMinMaxCompatibility() {
265265

266266
// All vertices are optional
267267
conf.set(AutoScalerOptions.UTILIZATION_TARGET, 0.6);
268-
conf.set(AutoScalerOptions.TARGET_UTILIZATION_BOUNDARY, 0.1);
268+
conf.set(AutoScalerOptions.UTILIZATION_TARGET_BOUNDARY, 0.1);
269269
var evaluated =
270270
Map.of(
271271
op1, evaluated(1, 70, 100),
@@ -278,7 +278,7 @@ op1, evaluated(1, 70, 100),
278278
ScalingExecutor.allChangedVerticesWithinUtilizationTarget(evaluated, Set.of(op2));
279279

280280
// Remove target boundary and use min max, should get the same result
281-
conf.removeConfig(AutoScalerOptions.TARGET_UTILIZATION_BOUNDARY);
281+
conf.removeConfig(AutoScalerOptions.UTILIZATION_TARGET_BOUNDARY);
282282
conf.set(AutoScalerOptions.UTILIZATION_MAX, 0.7);
283283
conf.set(AutoScalerOptions.UTILIZATION_MIN, 0.5);
284284
boolean minMaxOp1 =
@@ -291,15 +291,49 @@ op1, evaluated(1, 70, 100),
291291
// When the target boundary parameter is used,
292292
// but the min max parameter is also set,
293293
// the min max parameter shall prevail.
294-
conf.set(AutoScalerOptions.TARGET_UTILIZATION_BOUNDARY, 1.);
294+
conf.set(AutoScalerOptions.UTILIZATION_TARGET_BOUNDARY, 1.);
295295
conf.set(AutoScalerOptions.UTILIZATION_MAX, 0.7);
296296
conf.set(AutoScalerOptions.UTILIZATION_MIN, 0.3);
297-
298297
evaluated =
299298
Map.of(
300299
op1, evaluated(2, 150, 100),
301300
op2, evaluated(1, 85, 100));
301+
assertFalse(
302+
ScalingExecutor.allChangedVerticesWithinUtilizationTarget(
303+
evaluated, evaluated.keySet()));
304+
305+
// When the target boundary parameter is used,
306+
// but the max parameter is also set,
307+
conf.removeConfig(AutoScalerOptions.UTILIZATION_MIN);
308+
conf.set(AutoScalerOptions.UTILIZATION_TARGET_BOUNDARY, 1.);
309+
conf.set(AutoScalerOptions.UTILIZATION_TARGET, 0.5);
310+
conf.set(AutoScalerOptions.UTILIZATION_MAX, 0.6);
311+
312+
evaluated =
313+
Map.of(
314+
op1, evaluated(2, 100, 99999),
315+
op2, evaluated(1, 80, 99999));
316+
assertTrue(
317+
ScalingExecutor.allChangedVerticesWithinUtilizationTarget(
318+
evaluated, evaluated.keySet()));
319+
320+
evaluated = Map.of(op2, evaluated(1, 85, 100));
321+
assertFalse(
322+
ScalingExecutor.allChangedVerticesWithinUtilizationTarget(
323+
evaluated, evaluated.keySet()));
324+
325+
conf.removeConfig(AutoScalerOptions.UTILIZATION_MAX);
326+
conf.set(AutoScalerOptions.UTILIZATION_MIN, 0.3);
327+
328+
evaluated =
329+
Map.of(
330+
op1, evaluated(2, 80, 81),
331+
op2, evaluated(1, 100, 101));
332+
assertTrue(
333+
ScalingExecutor.allChangedVerticesWithinUtilizationTarget(
334+
evaluated, evaluated.keySet()));
302335

336+
evaluated = Map.of(op1, evaluated(1, 80, 79));
303337
assertFalse(
304338
ScalingExecutor.allChangedVerticesWithinUtilizationTarget(
305339
evaluated, evaluated.keySet()));

‎flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -611,7 +611,7 @@ public static Optional<String> validateAutoScalerFlinkConfiguration(
611611
validateNumber(flinkConfiguration, AutoScalerOptions.MAX_SCALE_UP_FACTOR, 0.0d),
612612
validateNumber(flinkConfiguration, UTILIZATION_TARGET, 0.0d, 1.0d),
613613
validateNumber(
614-
flinkConfiguration, AutoScalerOptions.TARGET_UTILIZATION_BOUNDARY, 0.0d),
614+
flinkConfiguration, AutoScalerOptions.UTILIZATION_TARGET_BOUNDARY, 0.0d),
615615
validateNumber(
616616
flinkConfiguration,
617617
UTILIZATION_MAX,

0 commit comments

Comments
 (0)