[FLINK-36527][autoscaler] Refactor JobVertexScaler to implement the parallelism adjustment logic in a separate component

Author: huyuanfeng

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

@@ -355,12 +355,9 @@ private boolean detectIneffectiveScaleUp(
      * But we limit newParallelism between parallelismLowerLimit and min(parallelismUpperLimit,
      * maxParallelism).
      *
-     * <p>Also, in order to ensure the data is evenly spread across subtasks, we try to adjust the
-     * parallelism for source and keyed vertex such that it divides the maxParallelism without a
-     * remainder.
-     *
-     * <p>This method also attempts to adjust the parallelism to ensure it aligns well with the
-     * number of source partitions if a vertex has a known source partition count.
+     * <p>Also, if we know the number of partitions or key groups corresponding to the current
+     * vertex, the parallelism will be adjusted accordingly. For specific logic, please refer to
+     * {@link ParallelismAdjuster}.
      */
     @VisibleForTesting
     protected static <KEY, Context extends JobAutoScalerContext<KEY>> int scale(
@@ -411,54 +408,15 @@ protected static <KEY, Context extends JobAutoScalerContext<KEY>> int scale(
 
         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 numKeyGroupsOrPartitions without a remainder => data is evenly spread across subtasks
-        for (int p = newParallelism; p <= upperBoundForAlignment; p++) {
-            if (numKeyGroupsOrPartitions % p == 0) {
-                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.
-        int p = newParallelism;
-        for (; p > 0; p--) {
-            if (numKeyGroupsOrPartitions / p > numKeyGroupsOrPartitions / newParallelism) {
-                if (numKeyGroupsOrPartitions % p != 0) {
-                    p++;
-                }
-                break;
-            }
-        }
 
-        p = Math.max(p, parallelismLowerLimit);
-        var message =
-                String.format(
-                        SCALE_LIMITED_MESSAGE_FORMAT,
-                        vertex,
-                        newParallelism,
-                        p,
-                        numKeyGroupsOrPartitions,
-                        upperBound,
-                        parallelismLowerLimit);
-        eventHandler.handleEvent(
+        return ParallelismAdjuster.adjustViaNumKeyGroupsOrPartitions(
+                vertex,
                 context,
-                AutoScalerEventHandler.Type.Warning,
-                SCALING_LIMITED,
-                message,
-                SCALING_LIMITED + vertex + (scaleFactor * currentParallelism),
-                context.getConfiguration().get(SCALING_EVENT_INTERVAL));
-        return p;
+                eventHandler,
+                numKeyGroupsOrPartitions,
+                newParallelism,
+                upperBound,
+                parallelismLowerLimit);
     }
 
     @VisibleForTesting

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

@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.autoscaler;
+
+import org.apache.flink.autoscaler.event.AutoScalerEventHandler;
+import org.apache.flink.autoscaler.topology.ShipStrategy;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+
+import static org.apache.flink.autoscaler.JobVertexScaler.SCALE_LIMITED_MESSAGE_FORMAT;
+import static org.apache.flink.autoscaler.JobVertexScaler.SCALING_LIMITED;
+import static org.apache.flink.autoscaler.config.AutoScalerOptions.SCALING_EVENT_INTERVAL;
+
+/**
+ * Component responsible adjusts the parallelism of a vertex.
+ *
+ * <p>When input vertex {@link ShipStrategy} is {@link ShipStrategy#HASH} or knows the number of
+ * current partitions of vertex. We hope to adjust the parallelism of the current vertex according
+ * to the number of key groups or partitions to achieve the goal of evenly distributing data among
+ * subtasks or maximizing utilization.
+ */
+public class ParallelismAdjuster {
+
+    public static <KEY, Context extends JobAutoScalerContext<KEY>>
+            int adjustViaNumKeyGroupsOrPartitions(
+                    JobVertexID vertex,
+                    Context context,
+                    AutoScalerEventHandler<KEY, Context> eventHandler,
+                    int numKeyGroupsOrPartitions,
+                    int newParallelism,
+                    int upperBound,
+                    int parallelismLowerLimit) {
+        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 numKeyGroupsOrPartitions without a remainder => data is evenly spread across subtasks
+        for (int p = newParallelism; p <= upperBoundForAlignment; p++) {
+            if (numKeyGroupsOrPartitions % p == 0) {
+                return p;
+            }
+        }
+
+        // When adjusting the parallelism after rounding up cannot
+        // find the right parallelism to meet requirements,
+        // Try to find the smallest parallelism that can satisfy the current consumption rate.
+        int p =
+                calculateMinimumParallelism(
+                        numKeyGroupsOrPartitions, newParallelism, parallelismLowerLimit);
+        var message =
+                String.format(
+                        SCALE_LIMITED_MESSAGE_FORMAT,
+                        vertex,
+                        newParallelism,
+                        p,
+                        numKeyGroupsOrPartitions,
+                        upperBound,
+                        parallelismLowerLimit);
+        eventHandler.handleEvent(
+                context,
+                AutoScalerEventHandler.Type.Warning,
+                SCALING_LIMITED,
+                message,
+                SCALING_LIMITED + vertex + newParallelism,
+                context.getConfiguration().get(SCALING_EVENT_INTERVAL));
+        return p;
+    }
+
+    private static int calculateMinimumParallelism(
+            int numKeyGroupsOrPartitions, int newParallelism, int parallelismLowerLimit) {
+        int p = newParallelism;
+        for (; p > 0; p--) {
+            if (numKeyGroupsOrPartitions / p > numKeyGroupsOrPartitions / newParallelism) {
+                if (numKeyGroupsOrPartitions % p != 0) {
+                    p++;
+                }
+                break;
+            }
+        }
+        p = Math.max(p, parallelismLowerLimit);
+        return p;
+    }
+}