[FLINK-36527][autoscaler] Introduce a parameter to support adopt a more radical strategy when source vertex or upstream shuffle is keyBy

Author: huyuanfeng

docs/layouts/shortcodes/generated/auto_scaler_configuration.html

@@ -188,6 +188,12 @@
             <td>Duration</td>
             <td>Time interval to resend the identical event</td>
         </tr>
+        <tr>
+            <td><h5>job.autoscaler.scaling.radical.enabled</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>If this option is enabled, The determination of parallelism will be more radical, which will maximize resource utilization, but may also cause data skew in some vertex.</td>
+        </tr>
         <tr>
             <td><h5>job.autoscaler.stabilization.interval</h5></td>
             <td style="word-wrap: break-word;">5 min</td>

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

@@ -416,21 +416,31 @@ protected static <KEY, Context extends JobAutoScalerContext<KEY>> int scale(
                         // Optimize the case where newParallelism <= maxParallelism / 2
                         newParallelism > numKeyGroupsOrPartitions / 2
                                 ? numKeyGroupsOrPartitions
-                                : numKeyGroupsOrPartitions / 2,
+                                : numKeyGroupsOrPartitions / 2 + numKeyGroupsOrPartitions % 2,
                         upperBound);
 
+        boolean scalingRadical =
+                context.getConfiguration().get(AutoScalerOptions.SCALING_RADICAL_ENABLED);
+
         // When the shuffle type of vertex inputs contains keyBy or vertex is a source,
         // we try to adjust the parallelism such that it divides
         // the numKeyGroupsOrPartitions without a remainder => data is evenly spread across subtasks
         for (int p = newParallelism; p <= upperBoundForAlignment; p++) {
-            if (numKeyGroupsOrPartitions % p == 0) {
+            if (numKeyGroupsOrPartitions % p == 0
+                    ||
+                    // When scaling radical is enabled, Try to find the smallest parallelism that
+                    // can satisfy the
+                    // current consumption rate.
+                    (scalingRadical
+                            && numKeyGroupsOrPartitions / p
+                                    < numKeyGroupsOrPartitions / newParallelism)) {
                 return p;
             }
         }
 
-        // When adjust the parallelism after rounding up cannot be evenly divided by
-        // numKeyGroupsOrPartitions, Try to find the smallest parallelism that can satisfy the
-        // current consumption rate.
+        // When adjust the parallelism after rounding up cannot be
+        // find the right degree of parallelism to meet requirements,
+        // Try to find the smallest parallelism that can satisfy the current consumption rate.
         int p = newParallelism;
         for (; p > 0; p--) {
             if (numKeyGroupsOrPartitions / p > numKeyGroupsOrPartitions / newParallelism) {

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

@@ -351,4 +351,13 @@ private static ConfigOptions.OptionBuilder autoScalerConfig(String key) {
                     .withFallbackKeys(oldOperatorConfigKey("quota.cpu"))
                     .withDescription(
                             "Quota of the CPU count. When scaling would go beyond this number the the scaling is not going to happen.");
+
+    public static final ConfigOption<Boolean> SCALING_RADICAL_ENABLED =
+            autoScalerConfig("scaling.radical.enabled")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withFallbackKeys(oldOperatorConfigKey("scaling.radical.enabled"))
+                    .withDescription(
+                            "If this option is enabled, The determination of parallelism will be more radical, which"
+                                    + " will maximize resource utilization, but may also cause data skew in some vertex.");
 }

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

@@ -379,6 +379,37 @@ public void testParallelismComputationWithAdjustment(
                         maxParallelism,
                         eventCollector,
                         context));
+
+        assertEquals(
+                32,
+                JobVertexScaler.scale(
+                        vertex,
+                        10,
+                        inputShipStrategies,
+                        0,
+                        128,
+                        2.5,
+                        minParallelism,
+                        maxParallelism,
+                        eventCollector,
+                        context));
+
+        // scaling.radical.enabled  = true
+        Configuration conf = context.getConfiguration();
+        conf.set(AutoScalerOptions.SCALING_RADICAL_ENABLED, true);
+        assertEquals(
+                26,
+                JobVertexScaler.scale(
+                        vertex,
+                        10,
+                        inputShipStrategies,
+                        0,
+                        128,
+                        2.5,
+                        minParallelism,
+                        maxParallelism,
+                        eventCollector,
+                        context));
     }
 
     @ParameterizedTest
@@ -1004,6 +1035,37 @@ public void testNumPartitionsAdjustment() {
                         parallelismUpperLimit,
                         eventCollector,
                         context));
+
+        assertEquals(
+                199,
+                JobVertexScaler.scale(
+                        vertex,
+                        24,
+                        List.of(),
+                        199,
+                        256,
+                        4,
+                        parallelismLowerLimit,
+                        parallelismUpperLimit,
+                        eventCollector,
+                        context));
+
+        // scaling.radical.enabled  = true
+        Configuration conf = context.getConfiguration();
+        conf.set(AutoScalerOptions.SCALING_RADICAL_ENABLED, true);
+        assertEquals(
+                100,
+                JobVertexScaler.scale(
+                        vertex,
+                        24,
+                        List.of(),
+                        199,
+                        256,
+                        4,
+                        parallelismLowerLimit,
+                        parallelismUpperLimit,
+                        eventCollector,
+                        context));
     }
 
     @Test