fixed

Author: huyuanfeng

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

@@ -72,7 +72,7 @@ public class JobVertexScaler<KEY, Context extends JobAutoScalerContext<KEY>> {
     @VisibleForTesting
     protected static final String SCALE_LIMITED_MESSAGE_FORMAT =
             "Scaling limited detected for %s (expected parallelism: %s, actual parallelism %s). "
-                    + "Scaling limited due to source partitions : %s，"
+                    + "Scaling limited due to numKeyGroupsOrPartitions : %s，"
                     + "upperBoundForAlignment(maxParallelism or parallelismUpperLimit): %s, parallelismLowerLimit: %s.";
 
     private Clock clock = Clock.system(ZoneId.systemDefault());
@@ -448,7 +448,7 @@ protected static <KEY, Context extends JobAutoScalerContext<KEY>> int scale(
                         vertex,
                         newParallelism,
                         p,
-                        numSourcePartitions,
+                        numKeyGroupsOrPartitions,
                         upperBound,
                         parallelismLowerLimit);
         eventHandler.handleEvent(

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

@@ -933,8 +933,8 @@ public void testSendingIneffectiveScalingEvents(Collection<ShipStrategy> inputSh
 
     @Test
     public void testNumPartitionsAdjustment() {
-        final int minParallelism = 1;
-        final int maxParallelism = Integer.MAX_VALUE;
+        final int parallelismLowerLimit = 1;
+        final int parallelismUpperLimit = Integer.MAX_VALUE;
         final var vertex = new JobVertexID();
 
         assertEquals(
@@ -946,8 +946,8 @@ public void testNumPartitionsAdjustment() {
                         15,
                         128,
                         0.4,
-                        minParallelism,
-                        maxParallelism,
+                        parallelismLowerLimit,
+                        parallelismUpperLimit,
                         eventCollector,
                         context));
         assertEquals(
@@ -959,8 +959,8 @@ public void testNumPartitionsAdjustment() {
                         15,
                         128,
                         1.2,
-                        minParallelism,
-                        maxParallelism,
+                        parallelismLowerLimit,
+                        parallelismUpperLimit,
                         eventCollector,
                         context));
         assertEquals(
@@ -972,8 +972,8 @@ public void testNumPartitionsAdjustment() {
                         35,
                         30,
                         0.9,
-                        minParallelism,
-                        maxParallelism,
+                        parallelismLowerLimit,
+                        parallelismUpperLimit,
                         eventCollector,
                         context));
 
@@ -987,7 +987,7 @@ public void testNumPartitionsAdjustment() {
                         30,
                         1.1,
                         20,
-                        maxParallelism,
+                        parallelismUpperLimit,
                         eventCollector,
                         context));
 
@@ -1000,8 +1000,8 @@ public void testNumPartitionsAdjustment() {
                         200,
                         128,
                         1.4,
-                        minParallelism,
-                        maxParallelism,
+                        parallelismLowerLimit,
+                        parallelismUpperLimit,
                         eventCollector,
                         context));
     }