Add a new NumKeyGroupsOrPartitionsParallelismAdjuster to make the logic clearer

Author: huyuanfeng

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

@@ -1,3 +1,20 @@
+/*
+ * 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.config.AutoScalerOptions;
@@ -34,23 +51,16 @@ public static <KEY, Context extends JobAutoScalerContext<KEY>> int adjust(
                 context.getConfiguration()
                         .get(AutoScalerOptions.SCALING_KEY_GROUP_PARTITIONS_ADJUST_MODE);
 
-        var upperBoundForAlignment =
-                Math.min(
-                        // Optimize the case where newParallelism <= maxParallelism / 2
-                        newParallelism > numKeyGroupsOrPartitions / 2
-                                ? numKeyGroupsOrPartitions
-                                : numKeyGroupsOrPartitions / 2 + numKeyGroupsOrPartitions % 2,
-                        upperBound);
+        var upperBoundForAlignment = Math.min(numKeyGroupsOrPartitions, 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
                     ||
-                    // When MAXIMIZE_UTILISATION is enabled, Try to find the smallest parallelism
-                    // that
-                    // can satisfy the current consumption rate.
+                    // When Mode is MAXIMIZE_UTILISATION , Try to find the smallest parallelism
+                    // that can satisfy the current consumption rate.
                     (mode == Mode.MAXIMIZE_UTILISATION
                             && numKeyGroupsOrPartitions / p
                                     < numKeyGroupsOrPartitions / newParallelism)) {

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

@@ -394,7 +394,6 @@ public void testParallelismComputationWithAdjustment(
                         eventCollector,
                         context));
 
-        // scaling.radical.enabled  = true
         Configuration conf = context.getConfiguration();
         conf.set(
                 AutoScalerOptions.SCALING_KEY_GROUP_PARTITIONS_ADJUST_MODE,
@@ -1052,7 +1051,6 @@ public void testNumPartitionsAdjustment() {
                         eventCollector,
                         context));
 
-        // scaling.radical.enabled  = true
         Configuration conf = context.getConfiguration();
         conf.set(
                 AutoScalerOptions.SCALING_KEY_GROUP_PARTITIONS_ADJUST_MODE,