[FLINK-39938] FLIP-586: Composable Parallelism Alignment Modes for Flink Autoscaler

Author: Dennis-Mircea

docs/content.zh/docs/operations/plugins.md

@@ -277,3 +277,89 @@ The following steps demonstrate how to develop and use a custom metric evaluator
        ```text
        o.a.f.a.s.AutoscalerUtils [INFO ] Discovered custom metric evaluator via ServiceLoader: org.apache.flink.autoscaler.custom.CustomEvaluator.
        ```
+
+## Custom Parallelism Alignment Modes
+
+The autoscaler aligns a vertex's computed target parallelism to the number of key groups (keyBy) or source partitions to reduce data skew. The built-in modes (`BALANCED`, `EVENLY_SPREAD`, `OFF`) are selected by name through `job.autoscaler.scaling.alignment.mode`. A custom alignment mode can also be provided as a plugin by implementing the `AlignmentMode` SPI. Custom modes are discovered through the [Plugins](https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/filesystems/plugins) mechanism when running inside the Kubernetes operator, and through the standard Java `ServiceLoader` mechanism when running with `flink-autoscaler-standalone`. In both cases the implementation class must be registered in `META-INF/services`.
+
+All `job.autoscaler.*` keys related to custom alignment modes also support the legacy `kubernetes.operator.`-prefixed form as a fallback (for example, `kubernetes.operator.job.autoscaler.scaling.alignment.mode`). The canonical key takes precedence on overlap.
+
+The following steps demonstrate how to develop and use a custom alignment mode.
+
+1. Implement `org.apache.flink.autoscaler.alignment.AlignmentMode`:
+    ```java
+    package org.apache.flink.autoscaler.alignment;
+
+    /** Custom alignment mode snapping the computed parallelism down to the nearest divisor. */
+    public class CustomAlignmentMode implements AlignmentMode {
+
+        /** Apply to every vertex, not only the source and keyBy vertices the built-ins handle. */
+        @Override
+        public boolean isApplicable(Context ctx) {
+            return true;
+        }
+
+        @Override
+        public int align(Context ctx) {
+            // Number of key groups, or source partitions for a partitioned source.
+            int n = ctx.getNumSourcePartitions() > 0
+                    ? ctx.getNumSourcePartitions()
+                    : ctx.getMaxParallelism();
+            // Optional mode-specific parameter, read from the per-mode configuration.
+            int minParallelism = ctx.getModeConfiguration().getInteger("min-parallelism", 1);
+            for (int p = ctx.getNewParallelism(); p >= minParallelism; p--) {
+                if (n % p == 0) {
+                    return p;
+                }
+            }
+            return ctx.getNewParallelism();
+        }
+    }
+    ```
+   The `Context` exposes the current and computed target parallelism, the number of key groups or source partitions, the input ship strategies, the `JobAutoScalerContext`, the per-vertex evaluated metrics, the job topology, and the prefix-stripped per-mode `Configuration` (`getModeConfiguration()`). The autoscaler calls `align` only when `isApplicable(Context)` returns true. That method defaults to keyBy (hash) vertices and to partitioned sources that report a partition count (Kafka and Pulsar do by default), and a custom mode can override it to widen the scope, for example to align custom partitioned vertices.
+
+2. Create the service definition file `org.apache.flink.autoscaler.alignment.AlignmentMode` in `META-INF/services`:
+    ```text
+    org.apache.flink.autoscaler.alignment.CustomAlignmentMode
+    ```
+
+3. Use the Maven tool to package the project and generate the custom alignment mode JAR.
+
+4. Select the custom mode by name and point it at your implementation class. Any other `job.autoscaler.scaling.alignment.mode.<name>.*` entries are passed to the mode (prefix-stripped) through `Context#getModeConfiguration()`:
+    ```yaml
+    job.autoscaler.scaling.alignment.mode: custom-mode
+    job.autoscaler.scaling.alignment.mode.custom-mode.class: org.apache.flink.autoscaler.alignment.CustomAlignmentMode
+    job.autoscaler.scaling.alignment.mode.custom-mode.min-parallelism: 4
+    ```
+   {{< hint info >}}
+   The `<name>` is any identifier you choose. `custom-mode` is used here, but `CUSTOM_MODE` or any other style works just as well. It is matched exactly, including case, so the value of `scaling.alignment.mode` and the `<name>` in `scaling.alignment.mode.<name>.class` must be identical. As a result, `CUSTOM_MODE` and `custom-mode` are two different modes, not aliases. The only reserved names are the built-ins (`BALANCED`, `EVENLY_SPREAD`, `OFF`), which always resolve to the built-in modes regardless of any configured class.
+   {{< /hint >}}
+
+5. Deploy the mode.
+
+    - **Operator deployment** - create a Dockerfile to build a custom image from the `apache/flink-kubernetes-operator` official image and copy the generated JAR to a custom alignment mode plugin directory under `/opt/flink/plugins` (the value of the `FLINK_PLUGINS_DIR` environment variable in the flink-kubernetes-operator helm chart). The structure of the custom alignment mode directory under `/opt/flink/plugins` is as follows:
+        ```text
+        /opt/flink/plugins
+            ├── custom-alignment-mode
+            │   ├── custom-alignment-mode.jar
+            └── ...
+        ```
+
+        With the custom alignment mode directory location, the Dockerfile is defined as follows:
+        ```shell script
+        FROM apache/flink-kubernetes-operator
+        ENV FLINK_PLUGINS_DIR=/opt/flink/plugins
+        ENV CUSTOM_ALIGNMENT_MODE_DIR=custom-alignment-mode
+        RUN mkdir $FLINK_PLUGINS_DIR/$CUSTOM_ALIGNMENT_MODE_DIR
+        COPY custom-alignment-mode.jar $FLINK_PLUGINS_DIR/$CUSTOM_ALIGNMENT_MODE_DIR/
+        ```
+
+        Install the flink-kubernetes-operator helm chart with the custom image and verify the `deploy/flink-kubernetes-operator` log has:
+        ```text
+        o.a.f.k.o.u.AutoscalerUtils [INFO ] Discovered custom alignment mode for autoscaler from plugin directory[/opt/flink/plugins]: org.apache.flink.autoscaler.alignment.CustomAlignmentMode.
+        ```
+
+    - **Standalone autoscaler** - simply place the custom alignment mode JAR on the classpath of the `flink-autoscaler-standalone` process. It will be picked up automatically via Java's `ServiceLoader` and discovery will be logged:
+        ```text
+        o.a.f.a.s.AutoscalerUtils [INFO ] Discovered custom alignment mode via ServiceLoader: org.apache.flink.autoscaler.alignment.CustomAlignmentMode.
+        ```

docs/content/docs/operations/plugins.md

@@ -277,3 +277,89 @@ The following steps demonstrate how to develop and use a custom metric evaluator
         ```text
         o.a.f.a.s.AutoscalerUtils [INFO ] Discovered custom metric evaluator via ServiceLoader: org.apache.flink.autoscaler.custom.CustomEvaluator.
         ```
+
+## Custom Parallelism Alignment Modes
+
+The autoscaler aligns a vertex's computed target parallelism to the number of key groups (keyBy) or source partitions to reduce data skew. The built-in modes (`BALANCED`, `EVENLY_SPREAD`, `OFF`) are selected by name through `job.autoscaler.scaling.alignment.mode`. A custom alignment mode can also be provided as a plugin by implementing the `AlignmentMode` SPI. Custom modes are discovered through the [Plugins](https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/filesystems/plugins) mechanism when running inside the Kubernetes operator, and through the standard Java `ServiceLoader` mechanism when running with `flink-autoscaler-standalone`. In both cases the implementation class must be registered in `META-INF/services`.
+
+All `job.autoscaler.*` keys related to custom alignment modes also support the legacy `kubernetes.operator.`-prefixed form as a fallback (for example, `kubernetes.operator.job.autoscaler.scaling.alignment.mode`). The canonical key takes precedence on overlap.
+
+The following steps demonstrate how to develop and use a custom alignment mode.
+
+1. Implement `org.apache.flink.autoscaler.alignment.AlignmentMode`:
+    ```java
+    package org.apache.flink.autoscaler.alignment;
+
+    /** Custom alignment mode snapping the computed parallelism down to the nearest divisor. */
+    public class CustomAlignmentMode implements AlignmentMode {
+
+        /** Apply to every vertex, not only the source and keyBy vertices the built-ins handle. */
+        @Override
+        public boolean isApplicable(Context ctx) {
+            return true;
+        }
+
+        @Override
+        public int align(Context ctx) {
+            // Number of key groups, or source partitions for a partitioned source.
+            int n = ctx.getNumSourcePartitions() > 0
+                    ? ctx.getNumSourcePartitions()
+                    : ctx.getMaxParallelism();
+            // Optional mode-specific parameter, read from the per-mode configuration.
+            int minParallelism = ctx.getModeConfiguration().getInteger("min-parallelism", 1);
+            for (int p = ctx.getNewParallelism(); p >= minParallelism; p--) {
+                if (n % p == 0) {
+                    return p;
+                }
+            }
+            return ctx.getNewParallelism();
+        }
+    }
+    ```
+   The `Context` exposes the current and computed target parallelism, the number of key groups or source partitions, the input ship strategies, the `JobAutoScalerContext`, the per-vertex evaluated metrics, the job topology, and the prefix-stripped per-mode `Configuration` (`getModeConfiguration()`). The autoscaler calls `align` only when `isApplicable(Context)` returns true. That method defaults to keyBy (hash) vertices and to partitioned sources that report a partition count (Kafka and Pulsar do by default), and a custom mode can override it to widen the scope, for example to align custom partitioned vertices.
+
+2. Create the service definition file `org.apache.flink.autoscaler.alignment.AlignmentMode` in `META-INF/services`:
+    ```text
+    org.apache.flink.autoscaler.alignment.CustomAlignmentMode
+    ```
+
+3. Use the Maven tool to package the project and generate the custom alignment mode JAR.
+
+4. Select the custom mode by name and point it at your implementation class. Any other `job.autoscaler.scaling.alignment.mode.<name>.*` entries are passed to the mode (prefix-stripped) through `Context#getModeConfiguration()`:
+    ```yaml
+    job.autoscaler.scaling.alignment.mode: custom-mode
+    job.autoscaler.scaling.alignment.mode.custom-mode.class: org.apache.flink.autoscaler.alignment.CustomAlignmentMode
+    job.autoscaler.scaling.alignment.mode.custom-mode.min-parallelism: 4
+    ```
+   {{< hint info >}}
+   The `<name>` is any identifier you choose. `custom-mode` is used here, but `CUSTOM_MODE` or any other style works just as well. It is matched exactly, including case, so the value of `scaling.alignment.mode` and the `<name>` in `scaling.alignment.mode.<name>.class` must be identical. As a result, `CUSTOM_MODE` and `custom-mode` are two different modes, not aliases. The only reserved names are the built-ins (`BALANCED`, `EVENLY_SPREAD`, `OFF`), which always resolve to the built-in modes regardless of any configured class.
+   {{< /hint >}}
+
+5. Deploy the mode.
+
+    - **Operator deployment** - create a Dockerfile to build a custom image from the `apache/flink-kubernetes-operator` official image and copy the generated JAR to a custom alignment mode plugin directory under `/opt/flink/plugins` (the value of the `FLINK_PLUGINS_DIR` environment variable in the flink-kubernetes-operator helm chart). The structure of the custom alignment mode directory under `/opt/flink/plugins` is as follows:
+        ```text
+        /opt/flink/plugins
+            ├── custom-alignment-mode
+            │   ├── custom-alignment-mode.jar
+            └── ...
+        ```
+
+        With the custom alignment mode directory location, the Dockerfile is defined as follows:
+        ```shell script
+        FROM apache/flink-kubernetes-operator
+        ENV FLINK_PLUGINS_DIR=/opt/flink/plugins
+        ENV CUSTOM_ALIGNMENT_MODE_DIR=custom-alignment-mode
+        RUN mkdir $FLINK_PLUGINS_DIR/$CUSTOM_ALIGNMENT_MODE_DIR
+        COPY custom-alignment-mode.jar $FLINK_PLUGINS_DIR/$CUSTOM_ALIGNMENT_MODE_DIR/
+        ```
+
+        Install the flink-kubernetes-operator helm chart with the custom image and verify the `deploy/flink-kubernetes-operator` log has:
+        ```text
+        o.a.f.k.o.u.AutoscalerUtils [INFO ] Discovered custom alignment mode for autoscaler from plugin directory[/opt/flink/plugins]: org.apache.flink.autoscaler.alignment.CustomAlignmentMode.
+        ```
+
+    - **Standalone autoscaler** - simply place the custom alignment mode JAR on the classpath of the `flink-autoscaler-standalone` process. It will be picked up automatically via Java's `ServiceLoader` and discovery will be logged:
+        ```text
+        o.a.f.a.s.AutoscalerUtils [INFO ] Discovered custom alignment mode via ServiceLoader: org.apache.flink.autoscaler.alignment.CustomAlignmentMode.
+        ```

docs/layouts/shortcodes/generated/auto_scaler_configuration.html

@@ -200,6 +200,24 @@
             <td>Double</td>
             <td>Max scale up factor. 2.0 means job can only be scaled up with 200% of the current parallelism.</td>
         </tr>
+        <tr>
+            <td><h5>job.autoscaler.scaling.alignment.mode</h5></td>
+            <td style="word-wrap: break-word;">"BALANCED"</td>
+            <td>String</td>
+            <td>How the autoscaler aligns the parallelism of source vertices and keyBy (hash) vertices to the number of key groups or source partitions. One of the built-in modes (BALANCED reduces per-subtask load above the target, EVENLY_SPREAD aligns to an exact divisor for even data distribution, OFF disables alignment), or the name of a custom alignment mode whose class is configured via 'job.autoscaler.scaling.alignment.mode.&lt;name&gt;.class'. Supersedes the deprecated scaling.key-group.partitions.adjust.mode.</td>
+        </tr>
+        <tr>
+            <td><h5>job.autoscaler.scaling.alignment.mode.&lt;name&gt;.&lt;parameter&gt;</h5></td>
+            <td style="word-wrap: break-word;">(none)</td>
+            <td>String</td>
+            <td>An arbitrary parameter passed to the custom alignment mode named &lt;name&gt;. All such parameters are made available to the mode (with the prefix stripped) through Context.getModeConfiguration().</td>
+        </tr>
+        <tr>
+            <td><h5>job.autoscaler.scaling.alignment.mode.&lt;name&gt;.class</h5></td>
+            <td style="word-wrap: break-word;">(none)</td>
+            <td>String</td>
+            <td>The fully-qualified class name of the custom alignment mode named &lt;name&gt;, discovered as a plugin and selectable via scaling.alignment.mode=&lt;name&gt;.</td>
+        </tr>
         <tr>
             <td><h5>job.autoscaler.scaling.effectiveness.detection.enabled</h5></td>
             <td style="word-wrap: break-word;">false</td>
@@ -224,12 +242,6 @@
             <td>Duration</td>
             <td>Time interval to resend the identical event</td>
         </tr>
-        <tr>
-            <td><h5>job.autoscaler.scaling.key-group.partitions.adjust.mode</h5></td>
-            <td style="word-wrap: break-word;">EVENLY_SPREAD</td>
-            <td><p>Enum</p></td>
-            <td>How to adjust the parallelism of Source vertex or upstream shuffle is keyBy<br /><br />Possible values:<ul><li>"EVENLY_SPREAD": This mode ensures that the parallelism adjustment attempts to evenly distribute data across subtasks. It is particularly effective for source vertices that are aware of partition counts or vertices after 'keyBy' operation. The goal is to have the number of key groups or partitions be divisible by the set parallelism, ensuring even data distribution and reducing data skew.</li><li>"MAXIMIZE_UTILISATION": This model is to maximize resource utilization. In this mode, an attempt is made to set the parallelism that meets the current consumption rate requirements. It is not enforced that the number of key groups or partitions is divisible by the parallelism.</li></ul></td>
-        </tr>
         <tr>
             <td><h5>job.autoscaler.stabilization.interval</h5></td>
             <td style="word-wrap: break-word;">5 min</td>

flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/StandaloneAutoscalerEntrypoint.java

@@ -26,6 +26,7 @@
 import org.apache.flink.autoscaler.RestApiMetricsCollector;
 import org.apache.flink.autoscaler.ScalingExecutor;
 import org.apache.flink.autoscaler.ScalingMetricEvaluator;
+import org.apache.flink.autoscaler.alignment.AlignmentMode;
 import org.apache.flink.autoscaler.event.AutoScalerEventHandler;
 import org.apache.flink.autoscaler.metrics.ScalingMetricsEvaluatorPlugin;
 import org.apache.flink.autoscaler.standalone.flinkcluster.FlinkClusterJobListFetcher;
@@ -94,10 +95,11 @@ JobAutoScaler<KEY, Context> createJobAutoscaler(
                     AutoScalerStateStore<KEY, Context> stateStore) {
         Collection<ScalingMetricsEvaluatorPlugin> customEvaluators =
                 AutoscalerUtils.discoverCustomEvaluators();
+        Collection<AlignmentMode> alignmentModes = AutoscalerUtils.discoverAlignmentModes();
         return new JobAutoScalerImpl<>(
                 new RestApiMetricsCollector<>(),
                 new ScalingMetricEvaluator(customEvaluators),
-                new ScalingExecutor<>(eventHandler, stateStore),
+                new ScalingExecutor<>(eventHandler, stateStore, null, alignmentModes),
                 eventHandler,
                 new RescaleApiScalingRealizer<>(eventHandler),
                 stateStore);

flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/utils/AutoscalerUtils.java

@@ -17,6 +17,7 @@
 
 package org.apache.flink.autoscaler.standalone.utils;
 
+import org.apache.flink.autoscaler.alignment.AlignmentMode;
 import org.apache.flink.autoscaler.metrics.ScalingMetricsEvaluatorPlugin;
 
 import org.slf4j.Logger;
@@ -52,4 +53,25 @@ public static Collection<ScalingMetricsEvaluatorPlugin> discoverCustomEvaluators
                         });
         return customEvaluators;
     }
+
+    /**
+     * Discovers custom alignment modes for the standalone autoscaler via Java's {@link
+     * ServiceLoader} mechanism. Implementations must be registered under {@code
+     * META-INF/services/org.apache.flink.autoscaler.alignment.AlignmentMode} on the classpath.
+     * Built-in modes are not plugins and are resolved by name, so they are not returned here.
+     *
+     * @return The list of discovered custom alignment modes.
+     */
+    public static Collection<AlignmentMode> discoverAlignmentModes() {
+        List<AlignmentMode> alignmentModes = new ArrayList<>();
+        ServiceLoader.load(AlignmentMode.class)
+                .forEach(
+                        alignmentMode -> {
+                            LOG.info(
+                                    "Discovered custom alignment mode via ServiceLoader: {}.",
+                                    alignmentMode.getClass().getName());
+                            alignmentModes.add(alignmentMode);
+                        });
+        return alignmentModes;
+    }
 }

flink-autoscaler-standalone/src/test/java/org/apache/flink/autoscaler/standalone/utils/AutoscalerUtilsTest.java

@@ -17,6 +17,7 @@
 
 package org.apache.flink.autoscaler.standalone.utils;
 
+import org.apache.flink.autoscaler.alignment.AlignmentMode;
 import org.apache.flink.autoscaler.metrics.ScalingMetricsEvaluatorPlugin;
 import org.apache.flink.autoscaler.metrics.TestCustomEvaluator;
 
@@ -40,4 +41,14 @@ void testDiscoverCustomEvaluators() {
                 .hasSize(1)
                 .hasOnlyElementsOfType(TestCustomEvaluator.class);
     }
+
+    @Test
+    void testDiscoverAlignmentModes() {
+        Collection<AlignmentMode> alignmentModes = AutoscalerUtils.discoverAlignmentModes();
+
+        assertThat(alignmentModes)
+                .as("Expected to discover the TestAlignmentMode registered under META-INF/services")
+                .hasSize(1)
+                .hasOnlyElementsOfType(TestAlignmentMode.class);
+    }
 }