This is an automated email from the ASF dual-hosted git repository.
gyfora pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git
The following commit(s) were added to refs/heads/main by this push:
new 44027b9db [FLINK-39555] FLIP-575: Scaling Executor Plugin SPI for
Flink Autoscaler
44027b9db is described below
commit 44027b9dbdd780cee1245d17844f3634b193b80a
Author: Dennis-Mircea Ciupitu <[email protected]>
AuthorDate: Fri Jun 19 19:28:05 2026 +0300
[FLINK-39555] FLIP-575: Scaling Executor Plugin SPI for Flink Autoscaler
---
docs/content.zh/docs/operations/plugins.md | 104 ++++
docs/content/docs/operations/plugins.md | 104 ++++
.../generated/auto_scaler_configuration.html | 18 +
.../standalone/StandaloneAutoscalerEntrypoint.java | 6 +-
.../standalone/utils/AutoscalerUtils.java | 25 +
.../standalone/utils/AutoscalerUtilsTest.java | 15 +
...g.apache.flink.autoscaler.ScalingExecutorPlugin | 15 +
.../apache/flink/autoscaler/ScalingExecutor.java | 185 +++++-
.../flink/autoscaler/ScalingExecutorPlugin.java | 120 ++++
.../flink/autoscaler/config/AutoScalerOptions.java | 125 +++-
.../autoscaler/metrics/ScalingHistoryUtils.java | 15 +
.../flink/autoscaler/ScalingExecutorTest.java | 630 ++++++++++++++++++++-
.../flink/autoscaler/TestScalingExecutor.java | 211 +++++++
.../flink/autoscaler/TestScalingExecutorTest.java | 308 ++++++++++
.../operator/autoscaler/AutoscalerFactory.java | 8 +-
.../kubernetes/operator/utils/AutoscalerUtils.java | 43 +-
.../src/test/assembly/test-plugins-assembly.xml | 1 +
.../operator/utils/AutoscalerUtilsTest.java | 59 ++
...g.apache.flink.autoscaler.ScalingExecutorPlugin | 15 +
19 files changed, 1965 insertions(+), 42 deletions(-)
diff --git a/docs/content.zh/docs/operations/plugins.md
b/docs/content.zh/docs/operations/plugins.md
index b5ae5e1a2..dafb1aace 100644
--- a/docs/content.zh/docs/operations/plugins.md
+++ b/docs/content.zh/docs/operations/plugins.md
@@ -277,3 +277,107 @@ 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 Scaling Executors
+
+`ScalingExecutorPlugin` is a pluggable component that allows users to
intercept, modify, or veto computed scaling decisions before they are applied.
Custom scaling executors 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 implementatio [...]
+
+For each scaling cycle, the autoscaler invokes every custom scaling executor
selected via the `job.autoscaler.scaling.custom-executors` configuration
option. Plugins are chained in priority order (lower `priority()` values
execute first) and each plugin receives the (possibly modified) output of the
previous one. A plugin can approve the decision unchanged, modify the
per-vertex `ScalingSummary` map, or veto the entire scaling operation by
returning an empty map. A veto is surfaced as a [...]
+
+All `job.autoscaler.*` keys related to custom scaling executors also support
the legacy `kubernetes.operator.`-prefixed form as a fallback (for example,
`kubernetes.operator.job.autoscaler.scaling.custom-executors`). The canonical
key takes precedence on overlap.
+
+The following steps demonstrate how to develop and use a custom scaling
executor.
+
+1. Implement the `ScalingExecutorPlugin` interface:
+ ```java
+ package org.apache.flink.autoscaler.custom;
+
+ import org.apache.flink.autoscaler.JobAutoScalerContext;
+ import org.apache.flink.autoscaler.ScalingExecutorPlugin;
+ import org.apache.flink.autoscaler.ScalingSummary;
+ import org.apache.flink.runtime.jobgraph.JobVertexID;
+
+ import java.util.Collections;
+ import java.util.Map;
+
+ /** Custom implementation of {@link ScalingExecutorPlugin} that caps
parallelism. */
+ public class CustomScalingExecutor<KEY, CTX extends
JobAutoScalerContext<KEY>>
+ implements ScalingExecutorPlugin<KEY, CTX> {
+
+ @Override
+ public int priority() {
+ // Lower values execute earlier in the chain. Default is 0.
+ return 0;
+ }
+
+ @Override
+ public Map<JobVertexID, ScalingSummary> apply(
+ Context<KEY, CTX> context, Map<JobVertexID, ScalingSummary>
scalingSummaries) {
+ // Read plugin-specific options from the prefix-stripped scoped
Configuration.
+ int maxParallelism =
+ context.getConfiguration().getInteger("max-parallelism",
Integer.MAX_VALUE);
+
+ // Example: veto the scaling operation if any vertex would exceed
the cap.
+ for (var summary : scalingSummaries.values()) {
+ if (summary.getNewParallelism() > maxParallelism) {
+ return Collections.emptyMap();
+ }
+ }
+ // Approve the decision unchanged.
+ return scalingSummaries;
+ }
+ }
+ ```
+
+ The `Context` object exposes the autoscaler context for the current job,
the evaluated scaling metrics, the job topology, and the prefix-stripped
per-instance `Configuration`.
+
+2. Create the service definition file
`org.apache.flink.autoscaler.ScalingExecutorPlugin` in `META-INF/services` with
the fully-qualified class name of your implementation:
+ ```text
+ org.apache.flink.autoscaler.custom.CustomScalingExecutor
+ ```
+
+3. Use the Maven tool to package the project and generate the custom scaling
executor JAR.
+
+4. Select one or more custom scaling executors via configuration. Each
instance whose implementation class FQN matches the configured
`job.autoscaler.scaling.custom-executor.<name>.class` value will be invoked,
and any other `job.autoscaler.scaling.custom-executor.<name>.*` entries are
passed to the plugin as plugin-specific configuration with the
`job.autoscaler.scaling.custom-executor.<instance>.` prefix stripped. The
configuration shape mirrors Flink's metric-reporter idiom: a list of [...]
+ ```yaml
+ job.autoscaler.scaling.custom-executors: cap,audit
+ job.autoscaler.scaling.custom-executor.cap.class:
org.apache.flink.autoscaler.custom.CustomScalingExecutor
+ job.autoscaler.scaling.custom-executor.cap.max-parallelism: 64
+ job.autoscaler.scaling.custom-executor.audit.class:
org.apache.flink.autoscaler.custom.AuditScalingExecutor
+ ```
+
+5. Deploy the custom scaling executor.
+
+ - **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 scaling executor 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
scaling executor directory under `/opt/flink/plugins` is as follows:
+ ```text
+ /opt/flink/plugins
+ ├── custom-scaling-executor
+ │ ├── custom-scaling-executor.jar
+ └── ...
+ ```
+
+ With the custom scaling executor directory location, the Dockerfile is
defined as follows:
+ ```shell script
+ FROM apache/flink-kubernetes-operator
+ ENV FLINK_PLUGINS_DIR=/opt/flink/plugins
+ ENV CUSTOM_SCALING_EXECUTOR_DIR=custom-scaling-executor
+ RUN mkdir $FLINK_PLUGINS_DIR/$CUSTOM_SCALING_EXECUTOR_DIR
+ COPY custom-scaling-executor.jar
$FLINK_PLUGINS_DIR/$CUSTOM_SCALING_EXECUTOR_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 scaling executor
for autoscaler from plugin directory[/opt/flink/plugins]:
org.apache.flink.autoscaler.custom.CustomScalingExecutor.
+ ```
+
+ - **Standalone autoscaler** - simply place the custom scaling executor 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.u.AutoscalerUtils [INFO ] Discovered custom scaling executor
via ServiceLoader: org.apache.flink.autoscaler.custom.CustomScalingExecutor.
+ ```
+
+ {{< hint info >}}
+ Once configured, on every reconciliation `ScalingExecutor` resolves the
chain against the per-job `Configuration`, sorts it by ascending `priority()`,
and emits an INFO log line listing the executor plugins that will be applied in
order. This is useful for verifying that all expected instances were picked up
and chained correctly:
+ ```text
+ o.a.f.a.ScalingExecutor [INFO ] Custom scaling executors resolved and
sorted by priority: [[name=cap,
class=org.apache.flink.autoscaler.custom.CustomScalingExecutor, priority=0],
[name=audit, class=org.apache.flink.autoscaler.custom.AuditScalingExecutor,
priority=100]]
+ ```
+ {{< /hint >}}
diff --git a/docs/content/docs/operations/plugins.md
b/docs/content/docs/operations/plugins.md
index 21ef284d4..31d19e387 100644
--- a/docs/content/docs/operations/plugins.md
+++ b/docs/content/docs/operations/plugins.md
@@ -277,3 +277,107 @@ 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 Scaling Executors
+
+`ScalingExecutorPlugin` is a pluggable component that allows users to
intercept, modify, or veto computed scaling decisions before they are applied.
Custom scaling executors 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 implementatio [...]
+
+For each scaling cycle, the autoscaler invokes every custom scaling executor
selected via the `job.autoscaler.scaling.custom-executors` configuration
option. Plugins are chained in priority order (lower `priority()` values
execute first) and each plugin receives the (possibly modified) output of the
previous one. A plugin can approve the decision unchanged, modify the
per-vertex `ScalingSummary` map, or veto the entire scaling operation by
returning an empty map. A veto is surfaced as a [...]
+
+All `job.autoscaler.*` keys related to custom scaling executors also support
the legacy `kubernetes.operator.`-prefixed form as a fallback (for example,
`kubernetes.operator.job.autoscaler.scaling.custom-executors`). The canonical
key takes precedence on overlap.
+
+The following steps demonstrate how to develop and use a custom scaling
executor.
+
+1. Implement the `ScalingExecutorPlugin` interface:
+ ```java
+ package org.apache.flink.autoscaler.custom;
+
+ import org.apache.flink.autoscaler.JobAutoScalerContext;
+ import org.apache.flink.autoscaler.ScalingExecutorPlugin;
+ import org.apache.flink.autoscaler.ScalingSummary;
+ import org.apache.flink.runtime.jobgraph.JobVertexID;
+
+ import java.util.Collections;
+ import java.util.Map;
+
+ /** Custom implementation of {@link ScalingExecutorPlugin} that caps
parallelism. */
+ public class CustomScalingExecutor<KEY, CTX extends
JobAutoScalerContext<KEY>>
+ implements ScalingExecutorPlugin<KEY, CTX> {
+
+ @Override
+ public int priority() {
+ // Lower values execute earlier in the chain. Default is 0.
+ return 0;
+ }
+
+ @Override
+ public Map<JobVertexID, ScalingSummary> apply(
+ Context<KEY, CTX> context, Map<JobVertexID, ScalingSummary>
scalingSummaries) {
+ // Read plugin-specific options from the prefix-stripped scoped
Configuration.
+ int maxParallelism =
+ context.getConfiguration().getInteger("max-parallelism",
Integer.MAX_VALUE);
+
+ // Example: veto the scaling operation if any vertex would exceed
the cap.
+ for (var summary : scalingSummaries.values()) {
+ if (summary.getNewParallelism() > maxParallelism) {
+ return Collections.emptyMap();
+ }
+ }
+ // Approve the decision unchanged.
+ return scalingSummaries;
+ }
+ }
+ ```
+
+ The `Context` object exposes the autoscaler context for the current job,
the evaluated scaling metrics, the job topology, and the prefix-stripped
per-instance `Configuration`.
+
+2. Create the service definition file
`org.apache.flink.autoscaler.ScalingExecutorPlugin` in `META-INF/services` with
the fully-qualified class name of your implementation:
+ ```text
+ org.apache.flink.autoscaler.custom.CustomScalingExecutor
+ ```
+
+3. Use the Maven tool to package the project and generate the custom scaling
executor JAR.
+
+4. Select one or more custom scaling executors via configuration. Each
instance whose implementation class FQN matches the configured
`job.autoscaler.scaling.custom-executor.<name>.class` value will be invoked,
and any other `job.autoscaler.scaling.custom-executor.<name>.*` entries are
passed to the plugin as plugin-specific configuration with the
`job.autoscaler.scaling.custom-executor.<instance>.` prefix stripped. The
configuration shape mirrors Flink's metric-reporter idiom: a list of [...]
+ ```yaml
+ job.autoscaler.scaling.custom-executors: cap,audit
+ job.autoscaler.scaling.custom-executor.cap.class:
org.apache.flink.autoscaler.custom.CustomScalingExecutor
+ job.autoscaler.scaling.custom-executor.cap.max-parallelism: 64
+ job.autoscaler.scaling.custom-executor.audit.class:
org.apache.flink.autoscaler.custom.AuditScalingExecutor
+ ```
+
+5. Deploy the custom scaling executor.
+
+ - **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 scaling executor 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
scaling executor directory under `/opt/flink/plugins` is as follows:
+ ```text
+ /opt/flink/plugins
+ ├── custom-scaling-executor
+ │ ├── custom-scaling-executor.jar
+ └── ...
+ ```
+
+ With the custom scaling executor directory location, the Dockerfile is
defined as follows:
+ ```shell script
+ FROM apache/flink-kubernetes-operator
+ ENV FLINK_PLUGINS_DIR=/opt/flink/plugins
+ ENV CUSTOM_SCALING_EXECUTOR_DIR=custom-scaling-executor
+ RUN mkdir $FLINK_PLUGINS_DIR/$CUSTOM_SCALING_EXECUTOR_DIR
+ COPY custom-scaling-executor.jar
$FLINK_PLUGINS_DIR/$CUSTOM_SCALING_EXECUTOR_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 scaling executor
for autoscaler from plugin directory[/opt/flink/plugins]:
org.apache.flink.autoscaler.custom.CustomScalingExecutor.
+ ```
+
+ - **Standalone autoscaler** - simply place the custom scaling executor 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.u.AutoscalerUtils [INFO ] Discovered custom scaling executor
via ServiceLoader: org.apache.flink.autoscaler.custom.CustomScalingExecutor.
+ ```
+
+ {{< hint info >}}
+ Once configured, on every reconciliation `ScalingExecutor` resolves the
chain against the per-job `Configuration`, sorts it by ascending `priority()`,
and emits an INFO log line listing the executor plugins that will be applied in
order. This is useful for verifying that all expected instances were picked up
and chained correctly:
+ ```text
+ o.a.f.a.ScalingExecutor [INFO ] Custom scaling executors resolved and
sorted by priority: [[name=cap,
class=org.apache.flink.autoscaler.custom.CustomScalingExecutor, priority=0],
[name=audit, class=org.apache.flink.autoscaler.custom.AuditScalingExecutor,
priority=100]]
+ ```
+ {{< /hint >}}
diff --git a/docs/layouts/shortcodes/generated/auto_scaler_configuration.html
b/docs/layouts/shortcodes/generated/auto_scaler_configuration.html
index 2844f5a2f..3d7ab51cd 100644
--- a/docs/layouts/shortcodes/generated/auto_scaler_configuration.html
+++ b/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.custom-executor.<name>.<parameter></h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Free-form, plugin-specific options for the scaling custom
executor instance referenced by '<name>' in
'job.autoscaler.scaling.custom-executors'. All keys with this prefix are passed
to the plugin at runtime with the
'job.autoscaler.scaling.custom-executor.<name>.' prefix stripped.</td>
+ </tr>
+ <tr>
+
<td><h5>job.autoscaler.scaling.custom-executor.<name>.class</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Fully-qualified class name of the implementation to
instantiate for the custom scaling executor instance <name> declared in
'job.autoscaler.scaling.custom-executors'. The class must be discovered via the
Plugin / ServiceLoader mechanism and registered under
META-INF/services/org.apache.flink.autoscaler.ScalingExecutorPlugin.</td>
+ </tr>
+ <tr>
+ <td><h5>job.autoscaler.scaling.custom-executors</h5></td>
+ <td style="word-wrap: break-word;"></td>
+ <td>List<String></td>
+ <td>List of named custom scaling executors instances to register
with the autoscaler. For each <name> entry, the implementation class to
instantiate must be set via
'job.autoscaler.scaling.custom-executor.<name>.class', and additional
per-instance options live under
'job.autoscaler.scaling.custom-executor.<name>.<parameter>'. Custom
executors are executed in ascending order of their
ScalingExecutorPlugin#priority() value, regardless of the order in whi [...]
+ </tr>
<tr>
<td><h5>job.autoscaler.scaling.effectiveness.detection.enabled</h5></td>
<td style="word-wrap: break-word;">false</td>
diff --git
a/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/StandaloneAutoscalerEntrypoint.java
b/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/StandaloneAutoscalerEntrypoint.java
index 443c296e4..cf3aa1e04 100644
---
a/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/StandaloneAutoscalerEntrypoint.java
+++
b/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/StandaloneAutoscalerEntrypoint.java
@@ -25,6 +25,7 @@ import org.apache.flink.autoscaler.JobAutoScalerContext;
import org.apache.flink.autoscaler.JobAutoScalerImpl;
import org.apache.flink.autoscaler.RestApiMetricsCollector;
import org.apache.flink.autoscaler.ScalingExecutor;
+import org.apache.flink.autoscaler.ScalingExecutorPlugin;
import org.apache.flink.autoscaler.ScalingMetricEvaluator;
import org.apache.flink.autoscaler.event.AutoScalerEventHandler;
import org.apache.flink.autoscaler.metrics.ScalingMetricsEvaluatorPlugin;
@@ -94,10 +95,13 @@ public class StandaloneAutoscalerEntrypoint {
AutoScalerStateStore<KEY, Context> stateStore) {
Collection<ScalingMetricsEvaluatorPlugin> customEvaluators =
AutoscalerUtils.discoverCustomEvaluators();
+ Collection<ScalingExecutorPlugin<KEY, Context>> customExecutors =
+ AutoscalerUtils.discoverCustomScalingExecutors();
+
return new JobAutoScalerImpl<>(
new RestApiMetricsCollector<>(),
new ScalingMetricEvaluator(customEvaluators),
- new ScalingExecutor<>(eventHandler, stateStore),
+ new ScalingExecutor<>(eventHandler, stateStore, null,
customExecutors),
eventHandler,
new RescaleApiScalingRealizer<>(eventHandler),
stateStore);
diff --git
a/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/utils/AutoscalerUtils.java
b/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/utils/AutoscalerUtils.java
index 39f4f0539..006f88147 100644
---
a/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/utils/AutoscalerUtils.java
+++
b/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/utils/AutoscalerUtils.java
@@ -17,6 +17,8 @@
package org.apache.flink.autoscaler.standalone.utils;
+import org.apache.flink.autoscaler.JobAutoScalerContext;
+import org.apache.flink.autoscaler.ScalingExecutorPlugin;
import org.apache.flink.autoscaler.metrics.ScalingMetricsEvaluatorPlugin;
import org.slf4j.Logger;
@@ -52,4 +54,27 @@ public class AutoscalerUtils {
});
return customEvaluators;
}
+
+ /**
+ * Discovers custom scaling executors for the standalone autoscaler via
Java's {@link
+ * ServiceLoader} mechanism. Implementations must be registered under
{@code
+ * META-INF/services/org.apache.flink.autoscaler.ScalingExecutorPlugin} on
the classpath.
+ *
+ * @return The list of discovered custom scaling executors.
+ */
+ @SuppressWarnings("unchecked")
+ public static <KEY, Context extends JobAutoScalerContext<KEY>>
+ Collection<ScalingExecutorPlugin<KEY, Context>>
discoverCustomScalingExecutors() {
+ List<ScalingExecutorPlugin<KEY, Context>> customScalingExecutors = new
ArrayList<>();
+ ServiceLoader.load(ScalingExecutorPlugin.class)
+ .forEach(
+ plugin -> {
+ LOG.info(
+ "Discovered custom scaling executor via
ServiceLoader: {}.",
+ plugin.getClass().getName());
+ customScalingExecutors.add(
+ (ScalingExecutorPlugin<KEY, Context>)
plugin);
+ });
+ return customScalingExecutors;
+ }
}
diff --git
a/flink-autoscaler-standalone/src/test/java/org/apache/flink/autoscaler/standalone/utils/AutoscalerUtilsTest.java
b/flink-autoscaler-standalone/src/test/java/org/apache/flink/autoscaler/standalone/utils/AutoscalerUtilsTest.java
index d720dbdac..8bc126300 100644
---
a/flink-autoscaler-standalone/src/test/java/org/apache/flink/autoscaler/standalone/utils/AutoscalerUtilsTest.java
+++
b/flink-autoscaler-standalone/src/test/java/org/apache/flink/autoscaler/standalone/utils/AutoscalerUtilsTest.java
@@ -17,6 +17,9 @@
package org.apache.flink.autoscaler.standalone.utils;
+import org.apache.flink.autoscaler.JobAutoScalerContext;
+import org.apache.flink.autoscaler.ScalingExecutorPlugin;
+import org.apache.flink.autoscaler.TestScalingExecutor;
import org.apache.flink.autoscaler.metrics.ScalingMetricsEvaluatorPlugin;
import org.apache.flink.autoscaler.metrics.TestCustomEvaluator;
@@ -40,4 +43,16 @@ class AutoscalerUtilsTest {
.hasSize(1)
.hasOnlyElementsOfType(TestCustomEvaluator.class);
}
+
+ @Test
+ void testDiscoverCustomScalingExecutors() {
+ Collection<ScalingExecutorPlugin<Object,
JobAutoScalerContext<Object>>> scalingExecutors =
+ AutoscalerUtils.discoverCustomScalingExecutors();
+
+ assertThat(scalingExecutors)
+ .as(
+ "Expected to discover the TestScalingExecutor
registered under META-INF/services")
+ .hasSize(1)
+ .hasOnlyElementsOfType(TestScalingExecutor.class);
+ }
}
diff --git
a/flink-autoscaler-standalone/src/test/resources/META-INF/services/org.apache.flink.autoscaler.ScalingExecutorPlugin
b/flink-autoscaler-standalone/src/test/resources/META-INF/services/org.apache.flink.autoscaler.ScalingExecutorPlugin
new file mode 100644
index 000000000..7cccfa18e
--- /dev/null
+++
b/flink-autoscaler-standalone/src/test/resources/META-INF/services/org.apache.flink.autoscaler.ScalingExecutorPlugin
@@ -0,0 +1,15 @@
+# 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.
+org.apache.flink.autoscaler.TestScalingExecutor
diff --git
a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingExecutor.java
b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingExecutor.java
index a9b04276a..dae93f788 100644
---
a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingExecutor.java
+++
b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingExecutor.java
@@ -44,11 +44,18 @@ import javax.annotation.Nullable;
import java.time.Clock;
import java.time.Duration;
import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Collection;
import java.util.Collections;
+import java.util.Comparator;
import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.SortedMap;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Collectors;
import static
org.apache.flink.autoscaler.config.AutoScalerOptions.EXCLUDED_PERIODS;
import static
org.apache.flink.autoscaler.config.AutoScalerOptions.SCALING_ENABLED;
@@ -57,6 +64,7 @@ import static
org.apache.flink.autoscaler.event.AutoScalerEventHandler.SCALING_E
import static
org.apache.flink.autoscaler.event.AutoScalerEventHandler.SCALING_SUMMARY_HEADER_SCALING_EXECUTION_DISABLED;
import static
org.apache.flink.autoscaler.event.AutoScalerEventHandler.SCALING_SUMMARY_HEADER_SCALING_EXECUTION_ENABLED;
import static
org.apache.flink.autoscaler.metrics.ScalingHistoryUtils.addToScalingHistoryAndStore;
+import static
org.apache.flink.autoscaler.metrics.ScalingHistoryUtils.copyScalingSummaries;
/** Class responsible for executing scaling decisions. */
public class ScalingExecutor<KEY, Context extends JobAutoScalerContext<KEY>> {
@@ -70,27 +78,42 @@ public class ScalingExecutor<KEY, Context extends
JobAutoScalerContext<KEY>> {
public static final String RESOURCE_QUOTA_REACHED_MESSAGE =
"Resource usage is above the allowed limit for scaling operations.
Please adjust the resource quota manually.";
+ public static final String SCALING_VETOED_REASON = "ScalingVetoed";
+
private static final Logger LOG =
LoggerFactory.getLogger(ScalingExecutor.class);
private final JobVertexScaler<KEY, Context> jobVertexScaler;
private final AutoScalerEventHandler<KEY, Context> autoScalerEventHandler;
private final AutoScalerStateStore<KEY, Context> autoScalerStateStore;
private final ResourceCheck resourceCheck;
+ private final Collection<ScalingExecutorPlugin<KEY, Context>>
customExecutors;
public ScalingExecutor(
AutoScalerEventHandler<KEY, Context> autoScalerEventHandler,
AutoScalerStateStore<KEY, Context> autoScalerStateStore) {
- this(autoScalerEventHandler, autoScalerStateStore, null);
+ this(autoScalerEventHandler, autoScalerStateStore, null,
Collections.emptyList());
}
public ScalingExecutor(
AutoScalerEventHandler<KEY, Context> autoScalerEventHandler,
AutoScalerStateStore<KEY, Context> autoScalerStateStore,
@Nullable ResourceCheck resourceCheck) {
+ this(autoScalerEventHandler, autoScalerStateStore, resourceCheck,
Collections.emptyList());
+ }
+
+ public ScalingExecutor(
+ AutoScalerEventHandler<KEY, Context> autoScalerEventHandler,
+ AutoScalerStateStore<KEY, Context> autoScalerStateStore,
+ @Nullable ResourceCheck resourceCheck,
+ Collection<ScalingExecutorPlugin<KEY, Context>> customExecutors) {
this.jobVertexScaler = new JobVertexScaler<>(autoScalerEventHandler);
this.autoScalerEventHandler = autoScalerEventHandler;
this.autoScalerStateStore = autoScalerStateStore;
this.resourceCheck = resourceCheck != null ? resourceCheck : new
NoopResourceCheck();
+ this.customExecutors =
+ customExecutors == null
+ ? Collections.emptyList()
+ : Collections.unmodifiableCollection(customExecutors);
}
public boolean scaleResource(
@@ -143,6 +166,23 @@ public class ScalingExecutor<KEY, Context extends
JobAutoScalerContext<KEY>> {
return false;
}
+ scalingSummaries =
+ applyCustomExecutors(
+ context,
+ memoryTuningEnabled ?
configOverrides.newConfigWithOverrides(conf) : conf,
+ evaluatedMetrics,
+ jobTopology,
+ scalingSummaries);
+ if (scalingSummaries == null || scalingSummaries.isEmpty()) {
+ return false;
+ }
+
+ autoScalerEventHandler.handleScalingEvent(
+ context,
+ scalingSummaries,
+ SCALING_SUMMARY_HEADER_SCALING_EXECUTION_ENABLED,
+ conf.get(SCALING_EVENT_INTERVAL));
+
addToScalingHistoryAndStore(
autoScalerStateStore, context, scalingHistory, now,
scalingSummaries);
@@ -435,6 +475,11 @@ public class ScalingExecutor<KEY, Context extends
JobAutoScalerContext<KEY>> {
Instant now) {
var scaleEnabled = conf.get(SCALING_ENABLED);
var isExcluded = CalendarUtils.inExcludedPeriods(conf, now);
+ if (scaleEnabled && !isExcluded) {
+ // Not blocked. The scaling report is emitted later, once the
decision is final
+ // (after memory tuning, the max-resource check, and the custom
scaling executors).
+ return false;
+ }
String message;
if (!scaleEnabled) {
message =
@@ -443,20 +488,150 @@ public class ScalingExecutor<KEY, Context extends
JobAutoScalerContext<KEY>> {
SCALING_EXECUTION_DISABLED_REASON,
SCALING_ENABLED.key(),
false);
- } else if (isExcluded) {
+ } else {
message =
SCALING_SUMMARY_HEADER_SCALING_EXECUTION_DISABLED
+ String.format(
SCALING_EXECUTION_DISABLED_REASON,
EXCLUDED_PERIODS.key(),
conf.get(EXCLUDED_PERIODS));
- } else {
- message = SCALING_SUMMARY_HEADER_SCALING_EXECUTION_ENABLED;
}
autoScalerEventHandler.handleScalingEvent(
context, scalingSummaries, message,
conf.get(SCALING_EVENT_INTERVAL));
+ return true;
+ }
- return !scaleEnabled || isExcluded;
+ /**
+ * Apply pluggable scaling custom executors (SPI extension point).
+ *
+ * <p>The chain to apply is resolved per reconciliation against the
per-job Configuration,
+ * mirroring custom metric evaluator plugin flow.
+ */
+ private Map<JobVertexID, ScalingSummary> applyCustomExecutors(
+ Context context,
+ Configuration conf,
+ EvaluatedMetrics evaluatedMetrics,
+ JobTopology jobTopology,
+ Map<JobVertexID, ScalingSummary> scalingSummaries) {
+ Collection<Map.Entry<String, ScalingExecutorPlugin<KEY, Context>>>
chain =
+ resolveCustomScalingExecutors(conf);
+ if (chain.isEmpty()) {
+ return scalingSummaries;
+ }
+ for (Map.Entry<String, ScalingExecutorPlugin<KEY, Context>> entry :
chain) {
+ var instanceName = entry.getKey();
+ var plugin = entry.getValue();
+ var pluginClassName = plugin.getClass().getName();
+ var pluginConfiguration =
+ AutoScalerOptions.customScalingExecutorConfiguration(conf,
instanceName);
+ var pluginContext =
+ new ScalingExecutorPlugin.Context<>(
+ context, pluginConfiguration, evaluatedMetrics,
jobTopology);
+ var previousSummaries = copyScalingSummaries(scalingSummaries);
+ scalingSummaries = plugin.apply(pluginContext, scalingSummaries);
+ if (scalingSummaries == null || scalingSummaries.isEmpty()) {
+ LOG.info(
+ "Scaling vetoed by scaling custom executor instance
'{}' (class={}).",
+ instanceName,
+ pluginClassName);
+ autoScalerEventHandler.handleEvent(
+ context,
+ AutoScalerEventHandler.Type.Warning,
+ SCALING_VETOED_REASON,
+ String.format(
+ "Scaling decision vetoed by scaling custom
executor instance '%s' (class=%s).",
+ instanceName, pluginClassName),
+ SCALING_VETOED_REASON + ":" + instanceName,
+ conf.get(SCALING_EVENT_INTERVAL));
+ return null;
+ }
+ if (!scalingSummaries.equals(previousSummaries)) {
+ LOG.info(
+ "Scaling summaries modified by scaling custom executor
instance '{}' (class={}): {} -> {}",
+ instanceName,
+ pluginClassName,
+ previousSummaries.values(),
+ scalingSummaries.values());
+ }
+ }
+ return scalingSummaries;
+ }
+
+ /**
+ * Resolves the ordered chain of scaling custom executor instances to
apply for the given
+ * per-reconciliation {@link Configuration}, against the cached discovered
SPI bag. Returns an
+ * ordered, possibly-empty {@link Collection} of {@code (instance-name,
plugin)} pairs sorted by
+ * ascending {@link ScalingExecutorPlugin#priority()}. The instance name
is the user-chosen
+ * {@code <name>} from {@code job.autoscaler.scaling.custom-executors} and
is required
+ * downstream for both the scoped per-instance {@link Configuration}
lookup and the approve/veto
+ * event {@code messageKey}, hence the {@code Map.Entry} pairing.
+ */
+ @VisibleForTesting
+ Collection<Map.Entry<String, ScalingExecutorPlugin<KEY, Context>>>
+ resolveCustomScalingExecutors(Configuration conf) {
+ List<String> instances =
conf.get(AutoScalerOptions.SCALING_CUSTOM_EXECUTORS);
+ if (instances == null || instances.isEmpty()) {
+ return Collections.emptyList();
+ }
+ List<Map.Entry<String, ScalingExecutorPlugin<KEY, Context>>> resolved
= new ArrayList<>();
+ Set<String> seenInstanceNames = new HashSet<>();
+ for (String instance : instances) {
+ if (!seenInstanceNames.add(instance)) {
+ LOG.warn(
+ "Custom scaling executor instance name '{}' is
configured more than once; ignoring duplicates.",
+ instance);
+ continue;
+ }
+ String classKey =
AutoScalerOptions.customScalingExecutorClassKey(instance);
+ String configuredClassName =
+
conf.get(AutoScalerOptions.customScalingExecutorClassOption(instance));
+ if (configuredClassName == null || configuredClassName.isBlank()) {
+ LOG.warn(
+ "Custom scaling executor instance '{}' is configured
in '{}' but no implementation class is "
+ + "set via '{}'. Ignoring.",
+ instance,
+ AutoScalerOptions.SCALING_CUSTOM_EXECUTORS.key(),
+ classKey);
+ continue;
+ }
+ ScalingExecutorPlugin<KEY, Context> match = null;
+ for (ScalingExecutorPlugin<KEY, Context> plugin : customExecutors)
{
+ if (plugin.getClass().getName().equals(configuredClassName)) {
+ match = plugin;
+ break;
+ }
+ }
+ if (match == null) {
+ LOG.warn(
+ "No registered scaling executor matches class '{}'
configured for instance '{}' via '{}'. Discovered executors: {}.",
+ configuredClassName,
+ instance,
+ classKey,
+ customExecutors.stream()
+ .map(p -> p.getClass().getName())
+ .collect(Collectors.toList()));
+ continue;
+ }
+ resolved.add(Map.entry(instance, match));
+ }
+ if (resolved.isEmpty()) {
+ return Collections.emptyList();
+ }
+ resolved.sort(Comparator.comparingInt(e -> e.getValue().priority()));
+ LOG.info(
+ "Custom scaling executors resolved and sorted by priority: {}",
+ resolved.stream()
+ .map(
+ e ->
+ "[name="
+ + e.getKey()
+ + ", class="
+ +
e.getValue().getClass().getName()
+ + ", priority="
+ + e.getValue().priority()
+ + "]")
+ .collect(Collectors.toList()));
+ return Collections.unmodifiableList(resolved);
}
@VisibleForTesting
diff --git
a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingExecutorPlugin.java
b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingExecutorPlugin.java
new file mode 100644
index 000000000..d5a471392
--- /dev/null
+++
b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingExecutorPlugin.java
@@ -0,0 +1,120 @@
+/*
+ * 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.metrics.EvaluatedMetrics;
+import org.apache.flink.autoscaler.topology.JobTopology;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+
+import lombok.Value;
+
+import java.util.Collections;
+import java.util.Map;
+
+/**
+ * A pluggable plugin that allows users to intercept, modify, or reject
computed scaling decisions
+ * before they are applied. Implementations are invoked in the {@link
ScalingExecutor} after scaling
+ * summaries have been computed and the blocked check has passed, but before
the actual application
+ * of parallelism overrides.
+ *
+ * <p>Multiple plugin implementations can be chained. Each plugin receives the
(possibly modified)
+ * output of the previous plugin. If any plugin returns an empty map, the
scaling operation is
+ * vetoed entirely.
+ *
+ * <p>This was introduced as part of <a
+ *
href="https://cwiki.apache.org/confluence/display/FLINK/FLIP-575%3A+Scaling+Executor+Plugin+SPI+for+Flink+Autoscaler";>FLIP-575:
+ * Scaling Executor Plugin SPI for Flink Autoscaler</a> and is complementary
to <a
+ *
href="https://cwiki.apache.org/confluence/display/FLINK/FLIP-514%3A+Custom+Evaluator+plugin+for+Flink+Autoscaler";>FLIP-514:
+ * Custom Evaluator plugin for Flink Autoscaler</a> which provides
extensibility at the metric
+ * evaluation layer.
+ *
+ * <p>Implementations are discovered via Java's {@link
java.util.ServiceLoader} mechanism. To
+ * register a custom plugin, add the fully qualified class name of the
implementation to {@code
+ * META-INF/services/org.apache.flink.autoscaler.ScalingExecutorPlugin}.
+ *
+ * @param <KEY> The job key.
+ * @param <CTX> Instance of {@link JobAutoScalerContext}.
+ */
+public interface ScalingExecutorPlugin<KEY, CTX extends
JobAutoScalerContext<KEY>> {
+
+ /**
+ * Returns the priority of this plugin in the chain. Plugins with lower
priority values are
+ * executed first. The default priority is 0. Plugins with equal priority
have no guaranteed
+ * relative ordering.
+ *
+ * <p>Example usage: a resource-gating plugin (priority -100) should run
before a
+ * parallelism-alignment plugin (priority 0), which should run before a
cost-cap plugin
+ * (priority 100).
+ *
+ * @return the priority value; lower values execute first.
+ */
+ default int priority() {
+ return 0;
+ }
+
+ /**
+ * Applies the plugin logic to the computed scaling summaries before they
are executed.
+ *
+ * <p>Implementations can:
+ *
+ * <ul>
+ * <li>Return the summaries unmodified to approve the scaling decision
as-is.
+ * <li>Return a modified map (e.g., with adjusted parallelism values) to
alter the decision.
+ * <li>Return an empty map (via {@link Collections#emptyMap()}) to
veto/reject the scaling
+ * operation entirely.
+ * </ul>
+ *
+ * @param context The plugin context wrapping the autoscaler context,
configuration, evaluated
+ * metrics, and job topology.
+ * @param scalingSummaries The computed scaling summaries keyed by vertex
ID. This map contains
+ * only vertices that require a parallelism change.
+ * @return The (possibly modified) scaling summaries to apply, or an empty
map to veto the
+ * scaling.
+ */
+ Map<JobVertexID, ScalingSummary> apply(
+ Context<KEY, CTX> context, Map<JobVertexID, ScalingSummary>
scalingSummaries);
+
+ /**
+ * Immutable context wrapping all inputs available to the scaling executor
plugin, providing
+ * access to the autoscaler context, configuration, evaluated metrics, and
job topology.
+ *
+ * @param <KEY> The job key.
+ * @param <CTX> Instance of {@link JobAutoScalerContext}.
+ */
+ @Value
+ class Context<KEY, CTX extends JobAutoScalerContext<KEY>> {
+
+ /** The autoscaler context for the current job. */
+ CTX autoScalerContext;
+
+ /**
+ * The prefix-stripped, per-plugin configuration for this invocation,
populated from {@code
+ * job.autoscaler.scaling.custom-executor.<name>.<parameter>} entries
(with the {@code
+ * kubernetes.operator.} legacy fallback honored). Never {@code null};
empty when the plugin
+ * has no per-instance options configured.
+ */
+ Configuration configuration;
+
+ /** The evaluated scaling metrics for all vertices and global metrics.
*/
+ EvaluatedMetrics evaluatedMetrics;
+
+ /** The job topology. */
+ JobTopology jobTopology;
+ }
+}
diff --git
a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/config/AutoScalerOptions.java
b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/config/AutoScalerOptions.java
index c1d723e2e..ef94231a1 100644
---
a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/config/AutoScalerOptions.java
+++
b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/config/AutoScalerOptions.java
@@ -34,6 +34,7 @@ public class AutoScalerOptions {
public static final String AUTOSCALER_CONF_PREFIX = "job.autoscaler.";
public static final String CUSTOM_EVALUATOR_CONF_PREFIX =
"metrics.custom-evaluator.";
public static final String CUSTOM_EVALUATOR_CLASS_SUFFIX = "class";
+ public static final String SCALING_CUSTOM_EXECUTOR_CONF_PREFIX =
"scaling.custom-executor.";
private static String oldOperatorConfigKey(String key) {
return OLD_K8S_OP_CONF_PREFIX + AUTOSCALER_CONF_PREFIX + key;
@@ -496,25 +497,117 @@ public class AutoScalerOptions {
}
/**
- * Builds a fresh, prefix-stripped {@link Configuration} that exposes the
per-instance options
- * configured for the scaling custom metric evaluator instance identified
by {@code
- * instanceName} under both the canonical ({@code
- * job.autoscaler.metrics.custom-evaluator.<name>.}) and the legacy ({@code
- * kubernetes.operator.job.autoscaler.metrics.custom-evaluator.<name>.})
namespaces. The legacy
- * prefix is applied first so that, on overlap, values written under the
canonical prefix take
- * precedence, matching the semantics of {@link
- * org.apache.flink.configuration.ConfigOption#withFallbackKeys} used
elsewhere in this class.
+ * The per-instance options configured for the custom metric evaluator
{@code
+ * customEvaluatorName}, prefix-stripped from the {@code
+ * job.autoscaler.metrics.custom-evaluator.<name>.} namespace. See {@link
+ * #prefixStrippedConfiguration(Configuration, String, String)} for the
canonical/legacy merge
+ * semantics.
*/
public static Configuration customEvaluatorConfiguration(
Configuration configuration, String customEvaluatorName) {
- String canonicalPrefix =
- AUTOSCALER_CONF_PREFIX + CUSTOM_EVALUATOR_CONF_PREFIX +
customEvaluatorName + ".";
- String legacyPrefix =
- OLD_K8S_OP_CONF_PREFIX
- + AUTOSCALER_CONF_PREFIX
- + CUSTOM_EVALUATOR_CONF_PREFIX
- + customEvaluatorName
- + ".";
+ return prefixStrippedConfiguration(
+ configuration, CUSTOM_EVALUATOR_CONF_PREFIX,
customEvaluatorName);
+ }
+
+ public static final ConfigOption<List<String>> SCALING_CUSTOM_EXECUTORS =
+ autoScalerConfig("scaling.custom-executors")
+ .stringType()
+ .asList()
+ .defaultValues()
+
.withFallbackKeys(oldOperatorConfigKey("scaling.custom-executors"))
+ .withDescription(
+ "List of named custom scaling executors instances
to register with the autoscaler. "
+ + "For each <name> entry, the
implementation class to instantiate must be set via "
+ +
"'job.autoscaler.scaling.custom-executor.<name>.class', and additional
per-instance "
+ + "options live under
'job.autoscaler.scaling.custom-executor.<name>.<parameter>'. "
+ + "Custom executors are executed in
ascending order of their ScalingExecutorPlugin#priority() value, "
+ + "regardless of the order in which they
appear in this list.");
+
+ /**
+ * Documentation-only template option describing the per-instance class
FQN key. Not read at
+ * runtime. The {@link #customScalingExecutorClassKey(String)} is used to
derive the actual key
+ * for a given scaling executor instance.
+ */
+ @SuppressWarnings("unused")
+ public static final ConfigOption<String>
SCALING_CUSTOM_EXECUTOR_CLASS_TEMPLATE =
+ autoScalerConfig(SCALING_CUSTOM_EXECUTOR_CONF_PREFIX +
"<name>.class")
+ .stringType()
+ .noDefaultValue()
+ .withFallbackKeys(
+ oldOperatorConfigKey(
+ SCALING_CUSTOM_EXECUTOR_CONF_PREFIX +
"<name>.class"))
+ .withDescription(
+ "Fully-qualified class name of the implementation
to instantiate for the custom scaling executor "
+ + "instance <name> declared in
'job.autoscaler.scaling.custom-executors'. "
+ + "The class must be discovered via the
Plugin / ServiceLoader mechanism and registered "
+ + "under
META-INF/services/org.apache.flink.autoscaler.ScalingExecutorPlugin.");
+
+ /**
+ * Documentation-only template option describing the per-instance
free-form parameter keys. Not
+ * read at runtime. The actual scaling executor parameters are surfaced to
the scaling executor
+ * via {@link #customScalingExecutorConfiguration(Configuration, String)}.
+ */
+ @SuppressWarnings("unused")
+ public static final ConfigOption<String>
SCALING_CUSTOM_EXECUTOR_PARAMETER_TEMPLATE =
+ autoScalerConfig(SCALING_CUSTOM_EXECUTOR_CONF_PREFIX +
"<name>.<parameter>")
+ .stringType()
+ .noDefaultValue()
+ .withFallbackKeys(
+ oldOperatorConfigKey(
+ SCALING_CUSTOM_EXECUTOR_CONF_PREFIX +
"<name>.<parameter>"))
+ .withDescription(
+ "Free-form, plugin-specific options for the
scaling custom executor instance "
+ + "referenced by '<name>' in
'job.autoscaler.scaling.custom-executors'. All keys with "
+ + "this prefix are passed to the plugin at
runtime with the "
+ +
"'job.autoscaler.scaling.custom-executor.<name>.' prefix stripped.");
+
+ public static String customScalingExecutorClassKey(String instanceName) {
+ return AUTOSCALER_CONF_PREFIX
+ + SCALING_CUSTOM_EXECUTOR_CONF_PREFIX
+ + instanceName
+ + ".class";
+ }
+
+ public static String customScalingExecutorClassFallbackKey(String
instanceName) {
+ return OLD_K8S_OP_CONF_PREFIX +
customScalingExecutorClassKey(instanceName);
+ }
+
+ /**
+ * Builds the {@link ConfigOption} used to read the implementation class
FQN for the given
+ * custom scaling executor instance, including the legacy {@code
kubernetes.operator.} prefix as
+ * a fallback key (canonical key takes precedence on overlap).
+ */
+ public static ConfigOption<String> customScalingExecutorClassOption(String
instanceName) {
+ return ConfigOptions.key(customScalingExecutorClassKey(instanceName))
+ .stringType()
+ .noDefaultValue()
+
.withFallbackKeys(customScalingExecutorClassFallbackKey(instanceName));
+ }
+
+ /**
+ * The per-instance options configured for the custom scaling executor
{@code instanceName},
+ * prefix-stripped from the {@code
job.autoscaler.scaling.custom-executor.<name>.} namespace.
+ * See {@link #prefixStrippedConfiguration(Configuration, String, String)}
for the
+ * canonical/legacy merge semantics.
+ */
+ public static Configuration customScalingExecutorConfiguration(
+ Configuration configuration, String instanceName) {
+ return prefixStrippedConfiguration(
+ configuration, SCALING_CUSTOM_EXECUTOR_CONF_PREFIX,
instanceName);
+ }
+
+ /**
+ * Builds a fresh {@link Configuration} exposing the per-instance options
configured under
+ * {@code job.autoscaler.<confPrefix><name>.} (canonical) and {@code
+ * kubernetes.operator.job.autoscaler.<confPrefix><name>.} (legacy), with
that prefix stripped.
+ * The legacy prefix is applied first so that, on overlap, values written
under the canonical
+ * prefix take precedence, matching the semantics of {@link
+ * org.apache.flink.configuration.ConfigOption#withFallbackKeys} used
elsewhere in this class.
+ */
+ private static Configuration prefixStrippedConfiguration(
+ Configuration configuration, String confPrefix, String name) {
+ String canonicalPrefix = AUTOSCALER_CONF_PREFIX + confPrefix + name +
".";
+ String legacyPrefix = OLD_K8S_OP_CONF_PREFIX + canonicalPrefix;
Configuration merged = new Configuration();
configuration
.toMap()
diff --git
a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingHistoryUtils.java
b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingHistoryUtils.java
index 98d15220e..2c8772018 100644
---
a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingHistoryUtils.java
+++
b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingHistoryUtils.java
@@ -28,6 +28,7 @@ import org.apache.flink.runtime.jobgraph.JobVertexID;
import javax.annotation.Nonnull;
import java.time.Instant;
+import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.SortedMap;
@@ -129,4 +130,18 @@ public class ScalingHistoryUtils {
conf.get(AutoScalerOptions.VERTEX_SCALING_HISTORY_COUNT));
return scalingTracking;
}
+
+ public static Map<JobVertexID, ScalingSummary> copyScalingSummaries(
+ Map<JobVertexID, ScalingSummary> summaries) {
+ var copy = new HashMap<JobVertexID, ScalingSummary>(summaries.size());
+ summaries.forEach(
+ (vertex, summary) -> {
+ var snapshot = new ScalingSummary();
+
snapshot.setCurrentParallelism(summary.getCurrentParallelism());
+ snapshot.setNewParallelism(summary.getNewParallelism());
+ snapshot.setMetrics(summary.getMetrics());
+ copy.put(vertex, snapshot);
+ });
+ return copy;
+ }
}
diff --git
a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingExecutorTest.java
b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingExecutorTest.java
index ac07e1a32..454d2e4b8 100644
---
a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingExecutorTest.java
+++
b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingExecutorTest.java
@@ -23,7 +23,6 @@ import
org.apache.flink.autoscaler.event.TestingEventCollector;
import org.apache.flink.autoscaler.metrics.EvaluatedMetrics;
import org.apache.flink.autoscaler.metrics.EvaluatedScalingMetric;
import org.apache.flink.autoscaler.metrics.ScalingMetric;
-import org.apache.flink.autoscaler.resources.ResourceCheck;
import org.apache.flink.autoscaler.state.AutoScalerStateStore;
import org.apache.flink.autoscaler.state.InMemoryAutoScalerStateStore;
import org.apache.flink.autoscaler.topology.IOMetrics;
@@ -36,6 +35,7 @@ import org.apache.flink.runtime.instance.SlotSharingGroupId;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Nested;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
@@ -46,6 +46,8 @@ import java.time.Duration;
import java.time.Instant;
import java.time.ZoneId;
import java.time.ZonedDateTime;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -555,16 +557,9 @@ public class ScalingExecutorTest {
new ScalingExecutor<>(
eventCollector,
stateStore,
- new ResourceCheck() {
- @Override
- public boolean trySchedule(
- int currentInstances,
- int newInstances,
- double cpuPerInstance,
- MemorySize memoryPerInstance) {
- return false;
- }
- });
+ (currentInstances, newInstances, cpuPerInstance,
memoryPerInstance) ->
+ false,
+ Collections.emptyList());
// Scaling blocked due to unavailable resources
assertFalse(
@@ -1027,7 +1022,6 @@ public class ScalingExecutorTest {
jobTopology,
new DelayedScaleDown()));
if (quotaReached) {
- assertEquals("ScalingReport",
eventCollector.events.poll().getReason());
assertEquals("ResourceQuotaReached",
eventCollector.events.poll().getReason());
assertTrue(eventCollector.events.isEmpty());
}
@@ -1066,4 +1060,616 @@ public class ScalingExecutorTest {
e -> JobVertexID.fromHexString(e.getKey()),
e -> Integer.valueOf(e.getValue())));
}
+
+ @Nested
+ class ScalingExecutorPluginTest {
+
+ private JobVertexID source;
+ private JobVertexID sink;
+ private JobTopology jobTopology;
+ private EvaluatedMetrics metrics;
+
+ @BeforeEach
+ void setupFilter() {
+ source = new JobVertexID();
+ sink = new JobVertexID();
+ jobTopology =
+ new JobTopology(
+ new VertexInfo(source, Map.of(), 10, 1000, false,
null),
+ new VertexInfo(sink, Map.of(source, HASH), 10,
1000, false, null));
+ metrics =
+ new EvaluatedMetrics(
+ Map.of(source, evaluated(10, 110, 100), sink,
evaluated(10, 110, 100)),
+ dummyGlobalMetrics);
+ }
+
+ /**
+ * Registers the given plugins with the per-reconciliation resolver by
setting both {@code
+ * scaling.custom-executors} and the per-instance {@code
+ * scaling.custom-executor.<name>.class} keys on the shared {@code
conf}, then constructs a
+ * {@link ScalingExecutor} bound to those plugins as the discovered
SPI bag. The instance
+ * name is the plugin's FQN so that approve/veto event {@code
messageKey} assertions can
+ * stay phrased in terms of {@code plugin.getClass().getName()}.
+ */
+ @SafeVarargs
+ private ScalingExecutor<JobID, JobAutoScalerContext<JobID>>
executorWith(
+ ScalingExecutorPlugin<JobID, JobAutoScalerContext<JobID>>...
plugins) {
+ var instances = new ArrayList<String>();
+ for (var p : plugins) {
+ var name = p.getClass().getName();
+ instances.add(name);
+
conf.setString(AutoScalerOptions.customScalingExecutorClassKey(name), name);
+ }
+ conf.set(AutoScalerOptions.SCALING_CUSTOM_EXECUTORS, instances);
+ return new ScalingExecutor<JobID, JobAutoScalerContext<JobID>>(
+ eventCollector, stateStore, null, List.of(plugins));
+ }
+
+ @Test
+ void testFilterApprovesScaling() throws Exception {
+ // A filter that approves scaling (passes through unchanged).
+ ScalingExecutorPlugin<JobID, JobAutoScalerContext<JobID>>
approveFilter =
+ (ctx, summaries) -> summaries;
+
+ var executorWithFilter = executorWith(approveFilter);
+
+ var now = Instant.now();
+ assertTrue(
+ executorWithFilter.scaleResource(
+ context,
+ metrics,
+ new HashMap<>(),
+ new ScalingTracking(),
+ now,
+ jobTopology,
+ new DelayedScaleDown()));
+
+ assertFalse(stateStore.getParallelismOverrides(context).isEmpty());
+ }
+
+ @Test
+ void testFilterVetoesScaling() throws Exception {
+ // A filter that vetoes scaling.
+ ScalingExecutorPlugin<JobID, JobAutoScalerContext<JobID>>
vetoFilter =
+ (ctx, summaries) -> Collections.emptyMap();
+
+ var executorWithFilter = executorWith(vetoFilter);
+
+ var now = Instant.now();
+ assertFalse(
+ executorWithFilter.scaleResource(
+ context,
+ metrics,
+ new HashMap<>(),
+ new ScalingTracking(),
+ now,
+ jobTopology,
+ new DelayedScaleDown()));
+
+ assertTrue(stateStore.getParallelismOverrides(context).isEmpty());
+ }
+
+ @Test
+ void testFilterModifiesSummaries() throws Exception {
+ // A filter that removes one vertex from scaling summaries.
+ ScalingExecutorPlugin<JobID, JobAutoScalerContext<JobID>>
modifyFilter =
+ (ctx, summaries) -> {
+ // Keep only the first entry
+ var firstEntry =
summaries.entrySet().iterator().next();
+ return Map.of(firstEntry.getKey(),
firstEntry.getValue());
+ };
+
+ var executorWithFilter = executorWith(modifyFilter);
+
+ var now = Instant.now();
+ assertTrue(
+ executorWithFilter.scaleResource(
+ context,
+ metrics,
+ new HashMap<>(),
+ new ScalingTracking(),
+ now,
+ jobTopology,
+ new DelayedScaleDown()));
+
+ // Despite 2 vertices needing scaling, only 1 should be in
overrides
+ // because the filter removed one
+ var overrides = stateStore.getParallelismOverrides(context);
+ assertFalse(overrides.isEmpty());
+ }
+
+ @Test
+ void testFilterReturnsEmptyMapVetoesScaling() throws Exception {
+ // A filter that returns empty map (no vertices left to scale).
+ ScalingExecutorPlugin<JobID, JobAutoScalerContext<JobID>>
emptyMapFilter =
+ (ctx, summaries) -> Collections.emptyMap();
+
+ var executorWithFilter = executorWith(emptyMapFilter);
+
+ var now = Instant.now();
+ assertFalse(
+ executorWithFilter.scaleResource(
+ context,
+ metrics,
+ new HashMap<>(),
+ new ScalingTracking(),
+ now,
+ jobTopology,
+ new DelayedScaleDown()));
+
+ assertTrue(stateStore.getParallelismOverrides(context).isEmpty());
+ }
+
+ @Test
+ void testMultipleFiltersChained() throws Exception {
+ // First filter: approves scaling
+ ScalingExecutorPlugin<JobID, JobAutoScalerContext<JobID>>
approveFilter =
+ (ctx, summaries) -> summaries;
+
+ // Second filter: vetoes scaling
+ ScalingExecutorPlugin<JobID, JobAutoScalerContext<JobID>>
vetoFilter =
+ (ctx, summaries) -> Collections.emptyMap();
+
+ var executorWithFilters = executorWith(approveFilter, vetoFilter);
+
+ var now = Instant.now();
+ assertFalse(
+ executorWithFilters.scaleResource(
+ context,
+ metrics,
+ new HashMap<>(),
+ new ScalingTracking(),
+ now,
+ jobTopology,
+ new DelayedScaleDown()));
+
+ assertTrue(stateStore.getParallelismOverrides(context).isEmpty());
+ }
+
+ @Test
+ void testMultipleFiltersAllApprove() throws Exception {
+ // Both filters approve
+ ScalingExecutorPlugin<JobID, JobAutoScalerContext<JobID>> filter1 =
+ (ctx, summaries) -> summaries;
+ ScalingExecutorPlugin<JobID, JobAutoScalerContext<JobID>> filter2 =
+ (ctx, summaries) -> summaries;
+
+ var executorWithFilters = executorWith(filter1, filter2);
+
+ var now = Instant.now();
+ assertTrue(
+ executorWithFilters.scaleResource(
+ context,
+ metrics,
+ new HashMap<>(),
+ new ScalingTracking(),
+ now,
+ jobTopology,
+ new DelayedScaleDown()));
+
+ assertFalse(stateStore.getParallelismOverrides(context).isEmpty());
+ }
+
+ @Test
+ void testNoFiltersScalesNormally() throws Exception {
+ // Empty filter collection (default behavior)
+ var executorNoFilters =
+ new ScalingExecutor<>(
+ eventCollector, stateStore, null,
Collections.emptyList());
+
+ var now = Instant.now();
+ assertTrue(
+ executorNoFilters.scaleResource(
+ context,
+ metrics,
+ new HashMap<>(),
+ new ScalingTracking(),
+ now,
+ jobTopology,
+ new DelayedScaleDown()));
+
+ assertFalse(stateStore.getParallelismOverrides(context).isEmpty());
+ }
+
+ @Test
+ void testFilterReceivesCorrectContext() throws Exception {
+ // A filter that asserts it receives the expected context
parameters
+ var receivedContextHolder =
+ new Object() {
+ ScalingExecutorPlugin.Context<JobID,
JobAutoScalerContext<JobID>>
+ pluginContext;
+ Map<JobVertexID, ScalingSummary> summaries;
+ };
+
+ ScalingExecutorPlugin<JobID, JobAutoScalerContext<JobID>>
captureFilter =
+ (ctx, summaries) -> {
+ receivedContextHolder.pluginContext = ctx;
+ receivedContextHolder.summaries = summaries;
+ return summaries;
+ };
+
+ var executorWithFilter = executorWith(captureFilter);
+
+ var now = Instant.now();
+ executorWithFilter.scaleResource(
+ context,
+ metrics,
+ new HashMap<>(),
+ new ScalingTracking(),
+ now,
+ jobTopology,
+ new DelayedScaleDown());
+
+ assertThat(receivedContextHolder.pluginContext).isNotNull();
+
assertThat(receivedContextHolder.pluginContext.getAutoScalerContext())
+ .isSameAs(context);
+
assertThat(receivedContextHolder.pluginContext.getConfiguration()).isNotNull();
+
assertThat(receivedContextHolder.pluginContext.getEvaluatedMetrics()).isSameAs(metrics);
+
assertThat(receivedContextHolder.pluginContext.getJobTopology()).isSameAs(jobTopology);
+ assertThat(receivedContextHolder.summaries).isNotEmpty();
+ // Both vertices should be in the summaries since both are above
target
+ assertThat(receivedContextHolder.summaries).containsKey(source);
+ assertThat(receivedContextHolder.summaries).containsKey(sink);
+ }
+
+ @Test
+ void testFilterPriorityOrdering() throws Exception {
+ // Track execution order
+ var executionOrder = new ArrayList<String>();
+
+ // High priority filter (priority = -10, should execute first)
+ ScalingExecutorPlugin<JobID, JobAutoScalerContext<JobID>>
highPriorityFilter =
+ new ScalingExecutorPlugin<>() {
+ @Override
+ public int priority() {
+ return -10;
+ }
+
+ @Override
+ public Map<JobVertexID, ScalingSummary> apply(
+ ScalingExecutorPlugin.Context<JobID,
JobAutoScalerContext<JobID>>
+ ctx,
+ Map<JobVertexID, ScalingSummary> summaries) {
+ executionOrder.add("high");
+ return summaries;
+ }
+ };
+
+ // Default priority filter (priority = 0, should execute second)
+ ScalingExecutorPlugin<JobID, JobAutoScalerContext<JobID>>
defaultPriorityFilter =
+ new ScalingExecutorPlugin<>() {
+ @Override
+ public Map<JobVertexID, ScalingSummary> apply(
+ ScalingExecutorPlugin.Context<JobID,
JobAutoScalerContext<JobID>>
+ ctx,
+ Map<JobVertexID, ScalingSummary> summaries) {
+ executionOrder.add("default");
+ return summaries;
+ }
+ };
+
+ // Low priority filter (priority = 10, should execute last)
+ ScalingExecutorPlugin<JobID, JobAutoScalerContext<JobID>>
lowPriorityFilter =
+ new ScalingExecutorPlugin<>() {
+ @Override
+ public int priority() {
+ return 10;
+ }
+
+ @Override
+ public Map<JobVertexID, ScalingSummary> apply(
+ ScalingExecutorPlugin.Context<JobID,
JobAutoScalerContext<JobID>>
+ ctx,
+ Map<JobVertexID, ScalingSummary> summaries) {
+ executionOrder.add("low");
+ return summaries;
+ }
+ };
+
+ // Register all three plugins unsorted; the per-reconciliation
resolver inside
+ // ScalingExecutor sorts the resolved chain by ascending
+ // ScalingExecutorPlugin#priority().
+ var executorWithFilters =
+ executorWith(lowPriorityFilter, defaultPriorityFilter,
highPriorityFilter);
+
+ var now = Instant.now();
+ assertTrue(
+ executorWithFilters.scaleResource(
+ context,
+ metrics,
+ new HashMap<>(),
+ new ScalingTracking(),
+ now,
+ jobTopology,
+ new DelayedScaleDown()));
+
+ // Verify they executed in priority order: high (-10), default
(0), low (10)
+ assertThat(executionOrder).containsExactly("high", "default",
"low");
+ }
+
+ @Test
+ void testApprovingPluginAllowsScalingWithoutVetoEvent() throws
Exception {
+ ScalingExecutorPlugin<JobID, JobAutoScalerContext<JobID>>
approveFilter =
+ (ctx, summaries) -> summaries;
+
+ var executorWithFilter = executorWith(approveFilter);
+
+ assertTrue(
+ executorWithFilter.scaleResource(
+ context,
+ metrics,
+ new HashMap<>(),
+ new ScalingTracking(),
+ Instant.now(),
+ jobTopology,
+ new DelayedScaleDown()));
+
+
assertThat(findEventByReason(ScalingExecutor.SCALING_VETOED_REASON)).isEmpty();
+ }
+
+ @Test
+ void testVetoEventEmittedWhenPluginVetoes() throws Exception {
+ ScalingExecutorPlugin<JobID, JobAutoScalerContext<JobID>>
vetoFilter =
+ (ctx, summaries) -> Collections.emptyMap();
+
+ var executorWithFilter = executorWith(vetoFilter);
+
+ assertFalse(
+ executorWithFilter.scaleResource(
+ context,
+ metrics,
+ new HashMap<>(),
+ new ScalingTracking(),
+ Instant.now(),
+ jobTopology,
+ new DelayedScaleDown()));
+
+ var vetoEvent =
findEventByReason(ScalingExecutor.SCALING_VETOED_REASON).orElseThrow();
+
assertThat(vetoEvent.getMessage()).contains(vetoFilter.getClass().getName());
+ assertThat(vetoEvent.getMessageKey())
+ .isEqualTo(
+ ScalingExecutor.SCALING_VETOED_REASON
+ + ":"
+ + vetoFilter.getClass().getName());
+ }
+
+ @Test
+ void testVetoAfterApproveEmitsVetoEvent() throws Exception {
+ ScalingExecutorPlugin<JobID, JobAutoScalerContext<JobID>>
approveFilter =
+ (ctx, summaries) -> summaries;
+ ScalingExecutorPlugin<JobID, JobAutoScalerContext<JobID>>
vetoFilter =
+ (ctx, summaries) -> Collections.emptyMap();
+
+ var executorWithFilters = executorWith(approveFilter, vetoFilter);
+
+ assertFalse(
+ executorWithFilters.scaleResource(
+ context,
+ metrics,
+ new HashMap<>(),
+ new ScalingTracking(),
+ Instant.now(),
+ jobTopology,
+ new DelayedScaleDown()));
+
+ var vetoEvent =
findEventByReason(ScalingExecutor.SCALING_VETOED_REASON).orElseThrow();
+
assertThat(vetoEvent.getMessage()).contains(vetoFilter.getClass().getName());
+ }
+
+ @Test
+ void testNoPluginEventsEmittedWhenNoPluginsRegistered() throws
Exception {
+ var executorNoFilters =
+ new ScalingExecutor<>(
+ eventCollector, stateStore, null,
Collections.emptyList());
+
+ assertTrue(
+ executorNoFilters.scaleResource(
+ context,
+ metrics,
+ new HashMap<>(),
+ new ScalingTracking(),
+ Instant.now(),
+ jobTopology,
+ new DelayedScaleDown()));
+
+
assertThat(findEventByReason(ScalingExecutor.SCALING_VETOED_REASON)).isEmpty();
+ }
+
+ @Test
+ void testLegacyFallbackClassKeyResolvesPlugin() throws Exception {
+ ScalingExecutorPlugin<JobID, JobAutoScalerContext<JobID>>
vetoFilter =
+ (ctx, summaries) -> Collections.emptyMap();
+ var name = vetoFilter.getClass().getName();
+ conf.set(AutoScalerOptions.SCALING_CUSTOM_EXECUTORS,
List.of(name));
+
conf.setString(AutoScalerOptions.customScalingExecutorClassFallbackKey(name),
name);
+
+ var executorWithFilter =
+ new ScalingExecutor<>(eventCollector, stateStore, null,
List.of(vetoFilter));
+
+ assertFalse(
+ executorWithFilter.scaleResource(
+ context,
+ metrics,
+ new HashMap<>(),
+ new ScalingTracking(),
+ Instant.now(),
+ jobTopology,
+ new DelayedScaleDown()));
+
+ // Plugin actually ran -> veto event was emitted with the
configured instance name.
+ var vetoEvent =
findEventByReason(ScalingExecutor.SCALING_VETOED_REASON).orElseThrow();
+ assertThat(vetoEvent.getMessageKey())
+ .isEqualTo(ScalingExecutor.SCALING_VETOED_REASON + ":" +
name);
+ }
+
+ @Test
+ void testLegacyFallbackForCustomExecutorsList() throws Exception {
+ // The 'scaling.custom-executors' LIST itself is registered ONLY
via the legacy
+ // 'kubernetes.operator.'-prefixed key — the canonical key is
intentionally not set.
+ // The resolver must still pick up the configured instance via the
fallback.
+ ScalingExecutorPlugin<JobID, JobAutoScalerContext<JobID>>
vetoFilter =
+ (ctx, summaries) -> Collections.emptyMap();
+ var name = vetoFilter.getClass().getName();
+ // Set ONLY the legacy form of 'scaling.custom-executors'.
+ conf.setString(
+ AutoScalerOptions.OLD_K8S_OP_CONF_PREFIX
+ + AutoScalerOptions.AUTOSCALER_CONF_PREFIX
+ + "scaling.custom-executors",
+ name);
+ // Per-instance .class still set under the canonical namespace so
resolution succeeds.
+
conf.setString(AutoScalerOptions.customScalingExecutorClassKey(name), name);
+
+ var executorWithFilter =
+ new ScalingExecutor<JobID, JobAutoScalerContext<JobID>>(
+ eventCollector, stateStore, null,
List.of(vetoFilter));
+
+ assertFalse(
+ executorWithFilter.scaleResource(
+ context,
+ metrics,
+ new HashMap<>(),
+ new ScalingTracking(),
+ Instant.now(),
+ jobTopology,
+ new DelayedScaleDown()));
+
+ // Plugin actually ran -> veto event was emitted with the
configured instance name.
+ var vetoEvent =
findEventByReason(ScalingExecutor.SCALING_VETOED_REASON).orElseThrow();
+ assertThat(vetoEvent.getMessageKey())
+ .isEqualTo(ScalingExecutor.SCALING_VETOED_REASON + ":" +
name);
+ }
+
+ @Test
+ void testPluginReceivesScopedConfiguration() throws Exception {
+ // The resolver must pass to each plugin a per-instance scoped
Configuration whose
+ // keys have the 'job.autoscaler.scaling.custom-executor.<name>.'
prefix stripped.
+ var captured =
+ new java.util.concurrent.atomic.AtomicReference<
+ org.apache.flink.configuration.Configuration>();
+ ScalingExecutorPlugin<JobID, JobAutoScalerContext<JobID>>
capturingPlugin =
+ (ctx, summaries) -> {
+ captured.set(ctx.getConfiguration());
+ return summaries;
+ };
+
+ var executorWithFilter = executorWith(capturingPlugin);
+ var name = capturingPlugin.getClass().getName();
+ // Canonical-prefixed parameter keys.
+ conf.setString(
+ AutoScalerOptions.AUTOSCALER_CONF_PREFIX
+ +
AutoScalerOptions.SCALING_CUSTOM_EXECUTOR_CONF_PREFIX
+ + name
+ + ".threshold",
+ "0.85");
+ conf.setString(
+ AutoScalerOptions.AUTOSCALER_CONF_PREFIX
+ +
AutoScalerOptions.SCALING_CUSTOM_EXECUTOR_CONF_PREFIX
+ + name
+ + ".allow-scale-down",
+ "true");
+
+ assertTrue(
+ executorWithFilter.scaleResource(
+ context,
+ metrics,
+ new HashMap<>(),
+ new ScalingTracking(),
+ Instant.now(),
+ jobTopology,
+ new DelayedScaleDown()));
+
+ var scoped = captured.get();
+ assertThat(scoped).as("plugin received Configuration").isNotNull();
+ assertThat(scoped.toMap())
+ .as("plugin-specific keys exposed with prefix stripped")
+ .containsEntry("threshold", "0.85")
+ .containsEntry("allow-scale-down", "true");
+ assertThat(scoped.toMap()).containsKey("class");
+ }
+
+ @Test
+ void testPluginReceivesScopedConfigurationFromLegacyPrefix() throws
Exception {
+ var captured =
+ new java.util.concurrent.atomic.AtomicReference<
+ org.apache.flink.configuration.Configuration>();
+ ScalingExecutorPlugin<JobID, JobAutoScalerContext<JobID>>
capturingPlugin =
+ (ctx, summaries) -> {
+ captured.set(ctx.getConfiguration());
+ return summaries;
+ };
+
+ var executorWithFilter = executorWith(capturingPlugin);
+ var name = capturingPlugin.getClass().getName();
+ conf.setString(
+ AutoScalerOptions.OLD_K8S_OP_CONF_PREFIX
+ + AutoScalerOptions.AUTOSCALER_CONF_PREFIX
+ +
AutoScalerOptions.SCALING_CUSTOM_EXECUTOR_CONF_PREFIX
+ + name
+ + ".legacy-only",
+ "from-legacy");
+
+ assertTrue(
+ executorWithFilter.scaleResource(
+ context,
+ metrics,
+ new HashMap<>(),
+ new ScalingTracking(),
+ Instant.now(),
+ jobTopology,
+ new DelayedScaleDown()));
+
+ assertThat(captured.get().toMap())
+ .as("legacy-only key exposed to plugin with prefix
stripped")
+ .containsEntry("legacy-only", "from-legacy");
+ }
+
+ @Test
+ void testCanonicalConfigKeyOverridesLegacyOnOverlap() throws Exception
{
+ var captured =
+ new java.util.concurrent.atomic.AtomicReference<
+ org.apache.flink.configuration.Configuration>();
+ ScalingExecutorPlugin<JobID, JobAutoScalerContext<JobID>>
capturingPlugin =
+ (ctx, summaries) -> {
+ captured.set(ctx.getConfiguration());
+ return summaries;
+ };
+
+ var executorWithFilter = executorWith(capturingPlugin);
+ var name = capturingPlugin.getClass().getName();
+ conf.setString(
+ AutoScalerOptions.OLD_K8S_OP_CONF_PREFIX
+ + AutoScalerOptions.AUTOSCALER_CONF_PREFIX
+ +
AutoScalerOptions.SCALING_CUSTOM_EXECUTOR_CONF_PREFIX
+ + name
+ + ".threshold",
+ "legacy-value");
+ conf.setString(
+ AutoScalerOptions.AUTOSCALER_CONF_PREFIX
+ +
AutoScalerOptions.SCALING_CUSTOM_EXECUTOR_CONF_PREFIX
+ + name
+ + ".threshold",
+ "canonical-value");
+
+ assertTrue(
+ executorWithFilter.scaleResource(
+ context,
+ metrics,
+ new HashMap<>(),
+ new ScalingTracking(),
+ Instant.now(),
+ jobTopology,
+ new DelayedScaleDown()));
+
+ assertThat(captured.get().toMap())
+ .as("canonical prefix wins on overlap")
+ .containsEntry("threshold", "canonical-value");
+ }
+
+ private Optional<TestingEventCollector.Event<JobID,
JobAutoScalerContext<JobID>>>
+ findEventByReason(String reason) {
+ return eventCollector.events.stream()
+ .filter(e -> reason.equals(e.getReason()))
+ .findFirst();
+ }
+ }
}
diff --git
a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/TestScalingExecutor.java
b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/TestScalingExecutor.java
new file mode 100644
index 000000000..c121f01fe
--- /dev/null
+++
b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/TestScalingExecutor.java
@@ -0,0 +1,211 @@
+/*
+ * 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.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import
org.apache.flink.runtime.rest.messages.job.metrics.AggregateTaskManagerMetricsParameters;
+import
org.apache.flink.runtime.rest.messages.job.metrics.AggregatedMetricsResponseBody;
+import
org.apache.flink.runtime.rest.messages.job.metrics.AggregatedTaskManagerMetricsHeaders;
+import
org.apache.flink.runtime.rest.messages.job.metrics.MetricsAggregationParameter;
+import
org.apache.flink.runtime.rest.messages.job.metrics.MetricsFilterParameter;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * A test {@link ScalingExecutorPlugin} implementation that demonstrates a
CPU-aware, source-aligned
+ * scaling strategy:
+ *
+ * <ol>
+ * <li>Allow scaling only when the average TaskManager JVM CPU usage
(queried via REST API from
+ * {@code Status.JVM.CPU.Load}) is above a configurable threshold
(default 0.8, i.e. 80%).
+ * <li>Align all non-source vertex parallelism to match the source vertex's
new parallelism.
+ * <li>If the source vertex's parallelism did not increase, veto the scaling
entirely.
+ * </ol>
+ *
+ * <p>This plugin is intended for testing and demonstration purposes. In
production, deploy it as a
+ * plugin JAR with a {@code
META-INF/services/org.apache.flink.autoscaler.ScalingExecutorPlugin}
+ * file.
+ */
+public class TestScalingExecutor<KEY, Context extends
JobAutoScalerContext<KEY>>
+ implements ScalingExecutorPlugin<KEY, Context> {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(TestScalingExecutor.class);
+
+ /** The Flink TM metric name for JVM process CPU usage (0.0 to 1.0). */
+ private static final String TM_CPU_LOAD_METRIC = "Status.JVM.CPU.Load";
+
+ /** Default CPU load threshold. 0.8 = 80%. */
+ private static final double DEFAULT_CPU_THRESHOLD = 0.8;
+
+ private final double cpuThreshold;
+
+ public TestScalingExecutor() {
+ this(DEFAULT_CPU_THRESHOLD);
+ }
+
+ public TestScalingExecutor(double cpuThreshold) {
+ this.cpuThreshold = cpuThreshold;
+ }
+
+ @Override
+ public Map<JobVertexID, ScalingSummary> apply(
+ ScalingExecutorPlugin.Context<KEY, Context> context,
+ Map<JobVertexID, ScalingSummary> scalingSummaries) {
+
+ // Step 1: Query the avg JVM CPU load across all TaskManagers via REST
API.
+ double avgCpuLoad;
+ try {
+ avgCpuLoad = queryAverageTmCpuLoad(context.getAutoScalerContext());
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to query TaskManager CPU load via REST API.
Vetoing scaling as a"
+ + " safety measure.",
+ e);
+ return Collections.emptyMap();
+ }
+
+ if (avgCpuLoad < cpuThreshold) {
+ LOG.info(
+ "Average TM CPU load ({}) is below threshold ({}). Vetoing
scaling.",
+ avgCpuLoad,
+ cpuThreshold);
+ return Collections.emptyMap();
+ }
+ LOG.info(
+ "Average TM CPU load ({}) is above threshold ({}). Proceeding
with scaling.",
+ avgCpuLoad,
+ cpuThreshold);
+
+ // Step 2: Find the source vertex and its new parallelism.
+ JobVertexID sourceVertex = null;
+ ScalingSummary sourceSummary = null;
+ for (Map.Entry<JobVertexID, ScalingSummary> entry :
scalingSummaries.entrySet()) {
+ if (context.getJobTopology().isSource(entry.getKey())) {
+ sourceVertex = entry.getKey();
+ sourceSummary = entry.getValue();
+ break;
+ }
+ }
+
+ // Step 3: If source is not in scaling summaries or didn't scale up,
veto.
+ if (sourceSummary == null) {
+ LOG.info("No source vertex found in scaling summaries. Vetoing
scaling.");
+ return Collections.emptyMap();
+ }
+ if (!sourceSummary.isScaledUp()) {
+ LOG.info(
+ "Source vertex {} parallelism did not increase
(current={}, new={}). "
+ + "Vetoing scaling.",
+ sourceVertex,
+ sourceSummary.getCurrentParallelism(),
+ sourceSummary.getNewParallelism());
+ return Collections.emptyMap();
+ }
+ int sourceNewParallelism = sourceSummary.getNewParallelism();
+ LOG.info(
+ "Source vertex {} scaled up to parallelism {}. "
+ + "Aligning all other vertices to source parallelism.",
+ sourceVertex,
+ sourceNewParallelism);
+
+ // Step 4: Align all non-source vertices to the source's new
parallelism.
+ var alignedSummaries = new HashMap<JobVertexID, ScalingSummary>();
+ for (Map.Entry<JobVertexID, ScalingSummary> entry :
scalingSummaries.entrySet()) {
+ var vertexId = entry.getKey();
+ var summary = entry.getValue();
+
+ if (vertexId.equals(sourceVertex)) {
+ // Keep source as-is
+ alignedSummaries.put(vertexId, summary);
+ } else {
+ int currentParallelism = summary.getCurrentParallelism();
+ if (currentParallelism == sourceNewParallelism) {
+ // Already at source parallelism, skip this vertex
+ LOG.info(
+ "Vertex {} already at source parallelism {}.
Skipping.",
+ vertexId,
+ sourceNewParallelism);
+ continue;
+ }
+ // Create a new ScalingSummary aligned to source parallelism
+ alignedSummaries.put(
+ vertexId,
+ new ScalingSummary(
+ currentParallelism, sourceNewParallelism,
summary.getMetrics()));
+ LOG.info(
+ "Aligned vertex {} parallelism from {} -> {} (source
parallelism).",
+ vertexId,
+ summary.getNewParallelism(),
+ sourceNewParallelism);
+ }
+ }
+
+ if (alignedSummaries.isEmpty()) {
+ LOG.info("No vertices require scaling after alignment. Vetoing
scaling.");
+ return Collections.emptyMap();
+ }
+
+ return alignedSummaries;
+ }
+
+ /**
+ * Queries the average JVM CPU load across all TaskManagers using the
Flink REST API's
+ * aggregated TM metrics endpoint ({@code
/taskmanagers/metrics?get=Status.JVM.CPU.Load}).
+ *
+ * <p>The metric {@code Status.JVM.CPU.Load} reports the JVM process CPU
usage as a value
+ * between 0.0 (no CPU) and 1.0 (100% CPU).
+ *
+ * @return the average CPU load across all TaskManagers, or 0.0 if not
available.
+ */
+ private double queryAverageTmCpuLoad(Context context) throws Exception {
+ try (var restClient = context.getRestClusterClient()) {
+ var parameters = new AggregateTaskManagerMetricsParameters();
+ var queryParamIt = parameters.getQueryParameters().iterator();
+
+ // Set the metric filter to Status.JVM.CPU.Load
+ MetricsFilterParameter filterParameter = (MetricsFilterParameter)
queryParamIt.next();
+ filterParameter.resolve(List.of(TM_CPU_LOAD_METRIC));
+
+ // Request AVG aggregation
+ MetricsAggregationParameter aggregationParameter =
+ (MetricsAggregationParameter) queryParamIt.next();
+
aggregationParameter.resolve(List.of(MetricsAggregationParameter.AggregationMode.AVG));
+
+ AggregatedMetricsResponseBody response =
+ restClient
+ .sendRequest(
+
AggregatedTaskManagerMetricsHeaders.getInstance(),
+ parameters,
+ EmptyRequestBody.getInstance())
+ .get();
+
+ return response.getMetrics().stream()
+ .filter(m -> TM_CPU_LOAD_METRIC.equals(m.getId()))
+ .findFirst()
+ .map(m -> m.getAvg() != null ? m.getAvg() : 0.0)
+ .orElse(0.0);
+ }
+ }
+}
diff --git
a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/TestScalingExecutorTest.java
b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/TestScalingExecutorTest.java
new file mode 100644
index 000000000..db5ab7ae0
--- /dev/null
+++
b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/TestScalingExecutorTest.java
@@ -0,0 +1,308 @@
+/*
+ * 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.api.common.JobID;
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.autoscaler.metrics.EvaluatedMetrics;
+import org.apache.flink.autoscaler.metrics.EvaluatedScalingMetric;
+import org.apache.flink.autoscaler.metrics.ScalingMetric;
+import org.apache.flink.autoscaler.topology.JobTopology;
+import org.apache.flink.autoscaler.topology.VertexInfo;
+import org.apache.flink.client.program.rest.RestClusterClient;
+import org.apache.flink.configuration.Configuration;
+import
org.apache.flink.runtime.highavailability.nonha.standalone.StandaloneClientHAServices;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.metrics.MetricRegistry;
+import org.apache.flink.runtime.metrics.NoOpMetricRegistry;
+import org.apache.flink.runtime.metrics.groups.GenericMetricGroup;
+import org.apache.flink.runtime.rest.messages.EmptyResponseBody;
+import org.apache.flink.runtime.rest.messages.MessageHeaders;
+import org.apache.flink.runtime.rest.messages.MessageParameters;
+import org.apache.flink.runtime.rest.messages.RequestBody;
+import org.apache.flink.runtime.rest.messages.ResponseBody;
+import org.apache.flink.runtime.rest.messages.job.metrics.AggregatedMetric;
+import
org.apache.flink.runtime.rest.messages.job.metrics.AggregatedMetricsResponseBody;
+import
org.apache.flink.runtime.rest.messages.job.metrics.AggregatedTaskManagerMetricsHeaders;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+import static org.apache.flink.autoscaler.topology.ShipStrategy.HASH;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link TestScalingExecutor}. */
+class TestScalingExecutorTest {
+
+ private Configuration conf;
+ private JobVertexID source;
+ private JobVertexID sink;
+ private JobTopology jobTopology;
+
+ @BeforeEach
+ void setup() {
+ conf = new Configuration();
+ source = new JobVertexID();
+ sink = new JobVertexID();
+ jobTopology =
+ new JobTopology(
+ new VertexInfo(source, Map.of(), 10, 1000, false,
null),
+ new VertexInfo(sink, Map.of(source, HASH), 10, 1000,
false, null));
+ }
+
+ private ScalingExecutorPlugin.Context<JobID, JobAutoScalerContext<JobID>>
createPluginContext(
+ JobAutoScalerContext<JobID> autoScalerContext) {
+ return new ScalingExecutorPlugin.Context<>(
+ autoScalerContext, conf, createMetrics(), jobTopology);
+ }
+
+ @Test
+ void testCpuThresholdBoundary() {
+ var context = createContextWithCpuLoad(0.9);
+ var plugin = new TestScalingExecutor<JobID,
JobAutoScalerContext<JobID>>(0.9);
+
+ var summaries = createSummaries(source, 10, 20, sink, 10, 20);
+
+ var result = plugin.apply(createPluginContext(context), summaries);
+
+ assertThat(result).isNotEmpty();
+ }
+
+ @Test
+ void testDefaultThreshold() {
+ // Uses default threshold (0.8), CPU 0.7 -> below
+ var context = createContextWithCpuLoad(0.7);
+ var plugin = new TestScalingExecutor<JobID,
JobAutoScalerContext<JobID>>();
+
+ var summaries = createSummaries(source, 10, 20, sink, 10, 15);
+
+ var result = plugin.apply(createPluginContext(context), summaries);
+
+ assertThat(result).isEmpty();
+ }
+
+ @Test
+ void testVetoWhenRestClientFails() {
+ // REST client that throws an exception
+ var context = createContextWithFailingRestClient();
+ var plugin = new TestScalingExecutor<JobID,
JobAutoScalerContext<JobID>>(0.5);
+
+ var summaries = createSummaries(source, 10, 20, sink, 10, 15);
+
+ var result = plugin.apply(createPluginContext(context), summaries);
+
+ // Should veto as safety measure when REST query fails
+ assertThat(result).isEmpty();
+ }
+
+ @Test
+ void testVetoWhenAvgCpuBelowThreshold() {
+ // CPU 0.5 < threshold 0.8
+ var context = createContextWithCpuLoad(0.5);
+ var plugin = new TestScalingExecutor<JobID,
JobAutoScalerContext<JobID>>(0.8);
+
+ var summaries = createSummaries(source, 10, 20, sink, 10, 20);
+
+ var result = plugin.apply(createPluginContext(context), summaries);
+
+ assertThat(result).isEmpty();
+ }
+
+ @Test
+ void testAllowWhenAvgCpuAboveThreshold() {
+ // CPU 0.9 > threshold 0.8; source scales up
+ var context = createContextWithCpuLoad(0.9);
+ var plugin = new TestScalingExecutor<JobID,
JobAutoScalerContext<JobID>>(0.8);
+
+ var summaries = createSummaries(source, 10, 20, sink, 10, 15);
+
+ var result = plugin.apply(createPluginContext(context), summaries);
+
+ assertThat(result).isNotEmpty();
+ // Sink should be aligned to source's new parallelism (20)
+ assertThat(result.get(source).getNewParallelism()).isEqualTo(20);
+ assertThat(result.get(sink).getNewParallelism()).isEqualTo(20);
+ }
+
+ @Test
+ void testVetoWhenSourceNotScalingUp() {
+ // CPU is high but source scales DOWN (20 -> 10)
+ var context = createContextWithCpuLoad(0.9);
+ var plugin = new TestScalingExecutor<JobID,
JobAutoScalerContext<JobID>>(0.5);
+
+ var summaries = createSummaries(source, 20, 10, sink, 10, 20);
+
+ var result = plugin.apply(createPluginContext(context), summaries);
+
+ // Source didn't increase -> veto
+ assertThat(result).isEmpty();
+ }
+
+ @Test
+ void testVetoWhenSourceNotInSummaries() {
+ // CPU is high but source is not in summaries
+ var context = createContextWithCpuLoad(0.9);
+ var plugin = new TestScalingExecutor<JobID,
JobAutoScalerContext<JobID>>(0.5);
+
+ var summaries = new HashMap<JobVertexID, ScalingSummary>();
+ summaries.put(sink, new ScalingSummary(10, 20, createVertexMetrics()));
+
+ var result = plugin.apply(createPluginContext(context), summaries);
+
+ // No source in summaries -> veto
+ assertThat(result).isEmpty();
+ }
+
+ @Test
+ void testAlignSinkToSourceParallelism() {
+ // Source: 10 -> 30, Sink: 10 -> 15 (autoscaler computed differently)
+ var context = createContextWithCpuLoad(0.9);
+ var plugin = new TestScalingExecutor<JobID,
JobAutoScalerContext<JobID>>(0.5);
+
+ var summaries = createSummaries(source, 10, 30, sink, 10, 15);
+
+ var result = plugin.apply(createPluginContext(context), summaries);
+
+ assertThat(result).isNotEmpty();
+ assertThat(result).hasSize(2);
+ assertThat(result.get(source).getNewParallelism()).isEqualTo(30);
+ // Sink aligned to source's new parallelism
+ assertThat(result.get(sink).getNewParallelism()).isEqualTo(30);
+ }
+
+ @Test
+ void testSkipVertexAlreadyAtSourceParallelism() {
+ // Source: 10 -> 20, Sink: current=20, computed new=25
+ var context = createContextWithCpuLoad(0.9);
+ var plugin = new TestScalingExecutor<JobID,
JobAutoScalerContext<JobID>>(0.5);
+
+ var summaries = createSummaries(source, 10, 20, sink, 20, 25);
+
+ var result = plugin.apply(createPluginContext(context), summaries);
+
+ assertThat(result).isNotEmpty();
+ // Sink's current parallelism (20) == source's new parallelism (20) ->
sink skipped
+ assertThat(result).hasSize(1);
+ assertThat(result).containsKey(source);
+ }
+
+ /**
+ * Creates a {@link JobAutoScalerContext} with a mocked REST client that
returns the given CPU
+ * load as the avg value for {@code Status.JVM.CPU.Load}.
+ */
+ private JobAutoScalerContext<JobID> createContextWithCpuLoad(double
cpuLoad) {
+ MetricRegistry registry = NoOpMetricRegistry.INSTANCE;
+ GenericMetricGroup metricGroup = new GenericMetricGroup(registry,
null, "test");
+ JobID jobID = new JobID();
+ return new JobAutoScalerContext<>(
+ jobID,
+ jobID,
+ JobStatus.RUNNING,
+ conf,
+ metricGroup,
+ () -> createRestClientWithCpuMetric(cpuLoad));
+ }
+
+ /** Creates a context whose REST client throws on getRestClusterClient().
*/
+ private JobAutoScalerContext<JobID> createContextWithFailingRestClient() {
+ MetricRegistry registry = NoOpMetricRegistry.INSTANCE;
+ GenericMetricGroup metricGroup = new GenericMetricGroup(registry,
null, "test");
+ JobID jobID = new JobID();
+ return new JobAutoScalerContext<>(
+ jobID,
+ jobID,
+ JobStatus.RUNNING,
+ conf,
+ metricGroup,
+ () -> {
+ throw new RuntimeException("REST client unavailable");
+ });
+ }
+
+ /**
+ * Creates a {@link RestClusterClient} that returns an aggregated TM
metrics response with the
+ * given CPU load average for the {@code Status.JVM.CPU.Load} metric.
+ */
+ @SuppressWarnings("unchecked")
+ private RestClusterClient<String> createRestClientWithCpuMetric(double
cpuLoad) {
+ try {
+ return new RestClusterClient<>(
+ conf, "test-cluster", (c, e) -> new
StandaloneClientHAServices("localhost")) {
+ @Override
+ public <
+ M extends MessageHeaders<R, P, U>,
+ U extends MessageParameters,
+ R extends RequestBody,
+ P extends ResponseBody>
+ CompletableFuture<P> sendRequest(
+ M messageHeaders, U messageParameters, R
request) {
+ if (messageHeaders instanceof
AggregatedTaskManagerMetricsHeaders) {
+ var cpuMetric =
+ new AggregatedMetric(
+ "Status.JVM.CPU.Load",
+ cpuLoad, // min
+ cpuLoad, // max
+ cpuLoad, // avg
+ cpuLoad, // sum
+ null);
+ return (CompletableFuture<P>)
+ CompletableFuture.completedFuture(
+ new
AggregatedMetricsResponseBody(List.of(cpuMetric)));
+ }
+ return (CompletableFuture<P>)
+
CompletableFuture.completedFuture(EmptyResponseBody.getInstance());
+ }
+ };
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private Map<ScalingMetric, EvaluatedScalingMetric> createVertexMetrics() {
+ var metrics = new HashMap<ScalingMetric, EvaluatedScalingMetric>();
+ metrics.put(ScalingMetric.PARALLELISM, EvaluatedScalingMetric.of(10));
+ metrics.put(ScalingMetric.MAX_PARALLELISM,
EvaluatedScalingMetric.of(1000));
+ return metrics;
+ }
+
+ private EvaluatedMetrics createMetrics() {
+ var vertexMetrics = new HashMap<JobVertexID, Map<ScalingMetric,
EvaluatedScalingMetric>>();
+ vertexMetrics.put(source, createVertexMetrics());
+ vertexMetrics.put(sink, createVertexMetrics());
+ return new EvaluatedMetrics(vertexMetrics, Map.of());
+ }
+
+ private Map<JobVertexID, ScalingSummary> createSummaries(
+ JobVertexID sourceId,
+ int sourceCurrentP,
+ int sourceNewP,
+ JobVertexID sinkId,
+ int sinkCurrentP,
+ int sinkNewP) {
+ var metrics = createVertexMetrics();
+ var summaries = new HashMap<JobVertexID, ScalingSummary>();
+ summaries.put(sourceId, new ScalingSummary(sourceCurrentP, sourceNewP,
metrics));
+ summaries.put(sinkId, new ScalingSummary(sinkCurrentP, sinkNewP,
metrics));
+ return summaries;
+ }
+}
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/AutoscalerFactory.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/AutoscalerFactory.java
index 506187946..4d6bc5b1d 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/AutoscalerFactory.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/AutoscalerFactory.java
@@ -21,6 +21,7 @@ import org.apache.flink.autoscaler.JobAutoScaler;
import org.apache.flink.autoscaler.JobAutoScalerImpl;
import org.apache.flink.autoscaler.RestApiMetricsCollector;
import org.apache.flink.autoscaler.ScalingExecutor;
+import org.apache.flink.autoscaler.ScalingExecutorPlugin;
import org.apache.flink.autoscaler.ScalingMetricEvaluator;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.kubernetes.operator.autoscaler.state.ConfigMapStore;
@@ -32,6 +33,8 @@ import
org.apache.flink.kubernetes.operator.utils.EventRecorder;
import io.fabric8.kubernetes.client.KubernetesClient;
import io.javaoperatorsdk.operator.processing.event.ResourceID;
+import java.util.Collection;
+
/** The factory of {@link JobAutoScaler}. */
public class AutoscalerFactory {
@@ -44,11 +47,14 @@ public class AutoscalerFactory {
var stateStore = new KubernetesAutoScalerStateStore(new
ConfigMapStore(client));
var eventHandler = new KubernetesAutoScalerEventHandler(eventRecorder);
var customEvaluators = AutoscalerUtils.discoverCustomEvaluators(conf);
+ Collection<ScalingExecutorPlugin<ResourceID,
KubernetesJobAutoScalerContext>>
+ customExecutors =
AutoscalerUtils.discoverCustomScalingExecutors(conf);
return new JobAutoScalerImpl<>(
new RestApiMetricsCollector<>(),
new ScalingMetricEvaluator(customEvaluators),
- new ScalingExecutor<>(eventHandler, stateStore,
clusterResourceManager),
+ new ScalingExecutor<>(
+ eventHandler, stateStore, clusterResourceManager,
customExecutors),
eventHandler,
new KubernetesScalingRealizer(),
stateStore);
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/AutoscalerUtils.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/AutoscalerUtils.java
index a960994c9..cbe08a5f3 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/AutoscalerUtils.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/AutoscalerUtils.java
@@ -1,11 +1,10 @@
/*
- * 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
+ * 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
*
@@ -18,6 +17,8 @@
package org.apache.flink.kubernetes.operator.utils;
+import org.apache.flink.autoscaler.JobAutoScalerContext;
+import org.apache.flink.autoscaler.ScalingExecutorPlugin;
import org.apache.flink.autoscaler.metrics.ScalingMetricsEvaluatorPlugin;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
@@ -58,4 +59,32 @@ public class AutoscalerUtils {
});
return customEvaluators;
}
+
+ /**
+ * Discovers custom scaling executors for autoscaler.
+ *
+ * @param conf Base FlinkConfigManager configuration
+ * @return The list of discovered custom scaling executors.
+ */
+ @SuppressWarnings("unchecked")
+ public static <KEY, Context extends JobAutoScalerContext<KEY>>
+ Collection<ScalingExecutorPlugin<KEY, Context>>
discoverCustomScalingExecutors(
+ Configuration conf) {
+ List<ScalingExecutorPlugin<KEY, Context>> customScalingExecutors = new
ArrayList<>();
+ PluginUtils.createPluginManagerFromRootFolder(conf)
+ .load(ScalingExecutorPlugin.class)
+ .forEachRemaining(
+ customScalingExecutor -> {
+ LOG.info(
+ "Discovered custom scaling executor for
autoscaler from plugin directory[{}]: {}.",
+ System.getenv()
+ .getOrDefault(
+
ConfigConstants.ENV_FLINK_PLUGINS_DIR,
+
ConfigConstants.DEFAULT_FLINK_PLUGINS_DIRS),
+
customScalingExecutor.getClass().getName());
+ customScalingExecutors.add(
+ (ScalingExecutorPlugin<KEY, Context>)
customScalingExecutor);
+ });
+ return customScalingExecutors;
+ }
}
diff --git
a/flink-kubernetes-operator/src/test/assembly/test-plugins-assembly.xml
b/flink-kubernetes-operator/src/test/assembly/test-plugins-assembly.xml
index 67526784a..da7646cb7 100644
--- a/flink-kubernetes-operator/src/test/assembly/test-plugins-assembly.xml
+++ b/flink-kubernetes-operator/src/test/assembly/test-plugins-assembly.xml
@@ -33,6 +33,7 @@ under the License.
<include>org/apache/flink/kubernetes/operator/autoscaler/TestingAutoscaler.class</include>
<include>org/apache/flink/kubernetes/operator/mutator/TestMutator.java</include>
<include>org/apache/flink/autoscaler/metrics/TestCustomEvaluator.java</include>
+
<include>org/apache/flink/autoscaler/TestScalingExecutor.class</include>
</includes>
</fileSet>
<fileSet>
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/AutoscalerUtilsTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/AutoscalerUtilsTest.java
new file mode 100644
index 000000000..c7301269f
--- /dev/null
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/AutoscalerUtilsTest.java
@@ -0,0 +1,59 @@
+/*
+ * 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.kubernetes.operator.utils;
+
+import org.apache.flink.autoscaler.TestScalingExecutor;
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.kubernetes.operator.TestUtils;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test class for {@link AutoscalerUtils}. */
+public class AutoscalerUtilsTest {
+ @TempDir public Path temporaryFolder;
+
+ @Test
+ public void testDiscoverScalingExecutorPluginsLoadsRegisteredSpi() throws
IOException {
+ Map<String, String> originalEnv = System.getenv();
+ try {
+ Map<String, String> systemEnv = new HashMap<>(originalEnv);
+ systemEnv.put(
+ ConfigConstants.ENV_FLINK_PLUGINS_DIR,
+ TestUtils.getTestPluginsRootDir(temporaryFolder));
+ TestUtils.setEnv(systemEnv);
+
+ var discovered =
AutoscalerUtils.discoverCustomScalingExecutors(new Configuration());
+
+ assertThat(discovered)
+ .hasSize(1)
+ .extracting(p -> p.getClass().getName())
+ .contains(TestScalingExecutor.class.getName());
+ } finally {
+ TestUtils.setEnv(originalEnv);
+ }
+ }
+}
diff --git
a/flink-kubernetes-operator/src/test/resources/META-INF/services/org.apache.flink.autoscaler.ScalingExecutorPlugin
b/flink-kubernetes-operator/src/test/resources/META-INF/services/org.apache.flink.autoscaler.ScalingExecutorPlugin
new file mode 100644
index 000000000..7cccfa18e
--- /dev/null
+++
b/flink-kubernetes-operator/src/test/resources/META-INF/services/org.apache.flink.autoscaler.ScalingExecutorPlugin
@@ -0,0 +1,15 @@
+# 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.
+org.apache.flink.autoscaler.TestScalingExecutor
