fixed

Author: huyuanfeng

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

@@ -409,25 +409,19 @@ protected static <KEY, Context extends JobAutoScalerContext<KEY>> int scale(
             return newParallelism;
         }
 
-        final int numKeyGroupsOrPartitions;
-        final int upperBoundForAlignment;
-        if (numSourcePartitions <= 0) {
-            numKeyGroupsOrPartitions = maxParallelism;
-            upperBoundForAlignment =
-                    Math.min(
-                            // Optimize the case where newParallelism <= maxParallelism / 2
-                            newParallelism > maxParallelism / 2
-                                    ? maxParallelism
-                                    : maxParallelism / 2,
-                            upperBound);
-        } else {
-            numKeyGroupsOrPartitions = numSourcePartitions;
-            upperBoundForAlignment = Math.min(numSourcePartitions, upperBound);
-        }
+        var numKeyGroupsOrPartitions =
+                numSourcePartitions <= 0 ? maxParallelism : numSourcePartitions;
+        var upperBoundForAlignment =
+                Math.min(
+                        // Optimize the case where newParallelism <= maxParallelism / 2
+                        newParallelism > numKeyGroupsOrPartitions / 2
+                                ? numKeyGroupsOrPartitions
+                                : numKeyGroupsOrPartitions / 2,
+                        upperBound);
 
         // 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 adjustableMaxParallelism without a remainder => data is evenly spread across subtasks
+        // the numKeyGroupsOrPartitions without a remainder => data is evenly spread across subtasks
         for (int p = newParallelism; p <= upperBoundForAlignment; p++) {
             if (numKeyGroupsOrPartitions % p == 0) {
                 return p;

flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingMetric.java

@@ -56,6 +56,7 @@ public enum ScalingMetric {
 
     /** Source vertex partition count. */
     NUM_SOURCE_PARTITIONS(false),
+
     /** Upper boundary of the target data rate range. */
     SCALE_UP_RATE_THRESHOLD(false),
 

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

@@ -958,7 +958,7 @@ public void testNumPartitionsAdjustment() {
                         List.of(),
                         15,
                         128,
-                        0.8,
+                        1.2,
                         minParallelism,
                         maxParallelism,
                         eventCollector,