[FLINK-36836][Autoscaler] Remove the default value of max min, calculated by utilization.target.boundary

Author: huyuanfeng

docs/layouts/shortcodes/generated/auto_scaler_configuration.html

@@ -202,13 +202,13 @@
         </tr>
         <tr>
             <td><h5>job.autoscaler.utilization.max</h5></td>
-            <td style="word-wrap: break-word;">1.0</td>
+            <td style="word-wrap: break-word;">(none)</td>
             <td>Double</td>
             <td>Max vertex utilization</td>
         </tr>
         <tr>
             <td><h5>job.autoscaler.utilization.min</h5></td>
-            <td style="word-wrap: break-word;">0.4</td>
+            <td style="word-wrap: break-word;">(none)</td>
             <td>Double</td>
             <td>Min vertex utilization</td>
         </tr>

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

@@ -287,6 +287,7 @@ protected static void computeProcessingRateThresholds(
             Duration restartTime) {
 
         double targetUtilization = conf.get(UTILIZATION_TARGET);
+        double utilizationBoundary = conf.get(TARGET_UTILIZATION_BOUNDARY);
 
         double upperUtilization;
         double lowerUtilization;
@@ -297,16 +298,12 @@ protected static void computeProcessingRateThresholds(
             upperUtilization = 1.0;
             lowerUtilization = 0.0;
         } else {
-            if (conf.getOptional(UTILIZATION_MAX).isPresent()
-                    || conf.getOptional(UTILIZATION_MIN).isPresent()
-                    || conf.getOptional(TARGET_UTILIZATION_BOUNDARY).isEmpty()) {
-                upperUtilization = conf.get(UTILIZATION_MAX);
-                lowerUtilization = conf.get(UTILIZATION_MIN);
-            } else {
-                Double boundary = conf.get(TARGET_UTILIZATION_BOUNDARY);
-                upperUtilization = targetUtilization + boundary;
-                lowerUtilization = targetUtilization - boundary;
-            }
+            upperUtilization =
+                    conf.getOptional(UTILIZATION_MAX)
+                            .orElse(targetUtilization + utilizationBoundary);
+            lowerUtilization =
+                    conf.getOptional(UTILIZATION_MIN)
+                            .orElse(targetUtilization - utilizationBoundary);
         }
 
         double scaleUpThreshold =

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

@@ -104,21 +104,20 @@ private static ConfigOptions.OptionBuilder autoScalerConfig(String key) {
             autoScalerConfig("target.utilization.boundary")
                     .doubleType()
                     .defaultValue(0.3)
-                    .withFallbackKeys(oldOperatorConfigKey("target.utilization.boundary"))
                     .withDescription(
                             "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)]");
 
     public static final ConfigOption<Double> UTILIZATION_MAX =
             autoScalerConfig("utilization.max")
                     .doubleType()
-                    .defaultValue(1.)
+                    .noDefaultValue()
                     .withFallbackKeys(oldOperatorConfigKey("utilization.max"))
                     .withDescription("Max vertex utilization");
 
     public static final ConfigOption<Double> UTILIZATION_MIN =
             autoScalerConfig("utilization.min")
                     .doubleType()
-                    .defaultValue(0.4)
+                    .noDefaultValue()
                     .withFallbackKeys(oldOperatorConfigKey("utilization.min"))
                     .withDescription("Min vertex utilization");
 

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

@@ -294,12 +294,46 @@ op1, evaluated(1, 70, 100),
         conf.set(AutoScalerOptions.TARGET_UTILIZATION_BOUNDARY, 1.);
         conf.set(AutoScalerOptions.UTILIZATION_MAX, 0.7);
         conf.set(AutoScalerOptions.UTILIZATION_MIN, 0.3);
-
         evaluated =
                 Map.of(
                         op1, evaluated(2, 150, 100),
                         op2, evaluated(1, 85, 100));
+        assertFalse(
+                ScalingExecutor.allChangedVerticesWithinUtilizationTarget(
+                        evaluated, evaluated.keySet()));
+
+        // When the target boundary parameter is used,
+        // but the max parameter is also set,
+        conf.removeConfig(AutoScalerOptions.UTILIZATION_MIN);
+        conf.set(AutoScalerOptions.TARGET_UTILIZATION_BOUNDARY, 1.);
+        conf.set(AutoScalerOptions.UTILIZATION_TARGET, 0.5);
+        conf.set(AutoScalerOptions.UTILIZATION_MAX, 0.6);
+
+        evaluated =
+                Map.of(
+                        op1, evaluated(2, 100, 99999),
+                        op2, evaluated(1, 80, 99999));
+        assertTrue(
+                ScalingExecutor.allChangedVerticesWithinUtilizationTarget(
+                        evaluated, evaluated.keySet()));
+
+        evaluated = Map.of(op2, evaluated(1, 85, 100));
+        assertFalse(
+                ScalingExecutor.allChangedVerticesWithinUtilizationTarget(
+                        evaluated, evaluated.keySet()));
+
+        conf.removeConfig(AutoScalerOptions.UTILIZATION_MAX);
+        conf.set(AutoScalerOptions.UTILIZATION_MIN, 0.3);
+
+        evaluated =
+                Map.of(
+                        op1, evaluated(2, 80, 81),
+                        op2, evaluated(1, 100, 101));
+        assertTrue(
+                ScalingExecutor.allChangedVerticesWithinUtilizationTarget(
+                        evaluated, evaluated.keySet()));
 
+        evaluated = Map.of(op1, evaluated(1, 80, 79));
         assertFalse(
                 ScalingExecutor.allChangedVerticesWithinUtilizationTarget(
                         evaluated, evaluated.keySet()));