You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gy...@apache.org on 2026/07/07 16:16:44 UTC
(flink-kubernetes-operator) branch main updated: [FLINK-39959] Improve the autoscaler context structure and propagation
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 5dddcdc42 [FLINK-39959] Improve the autoscaler context structure and propagation
5dddcdc42 is described below
commit 5dddcdc42e6eedd6f55c05db6fd20384d1aee8e7
Author: Dennis-Mircea Ciupitu <de...@gmail.com>
AuthorDate: Tue Jul 7 19:16:39 2026 +0300
[FLINK-39959] Improve the autoscaler context structure and propagation
---
docs/content.zh/docs/operations/plugins.md | 28 ++--
docs/content/docs/operations/plugins.md | 28 ++--
.../generated/auto_scaler_configuration.html | 2 +-
.../standalone/StandaloneAutoscalerEntrypoint.java | 6 +-
.../standalone/utils/AutoscalerUtils.java | 7 +-
.../standalone/utils/AutoscalerUtilsTest.java | 3 +-
.../flink/autoscaler/JobAutoScalerContext.java | 147 ++++++++++++++++++++-
.../apache/flink/autoscaler/JobAutoScalerImpl.java | 78 ++---------
.../apache/flink/autoscaler/JobVertexScaler.java | 1 -
.../flink/autoscaler/RestApiMetricsCollector.java | 5 +
.../apache/flink/autoscaler/ScalingExecutor.java | 59 ++++++---
.../flink/autoscaler/ScalingExecutorPlugin.java | 49 +++----
.../flink/autoscaler/ScalingMetricCollector.java | 59 +++++++--
.../flink/autoscaler/ScalingMetricEvaluator.java | 133 +++++++++++++------
.../autoscaler/alignment/BuiltInAlignmentMode.java | 8 +-
.../alignment/KeyGroupOrPartitionsAdjustMode.java | 4 +-
.../autoscaler/alignment/ParallelismAligner.java | 28 ++--
.../alignment/ParallelismAlignmentMode.java | 80 ++++++-----
.../flink/autoscaler/config/AutoScalerOptions.java | 16 ++-
.../metrics/ScalingMetricsEvaluatorPlugin.java | 57 ++++----
.../autoscaler/AutoScalerCustomEvaluatorTest.java | 5 +-
.../flink/autoscaler/BacklogBasedScalingTest.java | 7 +-
.../autoscaler/DelayedScaleDownEndToEndTest.java | 9 +-
.../flink/autoscaler/JobAutoScalerImplTest.java | 20 +--
.../MetricsCollectionAndEvaluationTest.java | 117 ++++++++--------
.../autoscaler/RecommendedParallelismTest.java | 10 +-
.../autoscaler/RestApiMetricsCollectorTest.java | 13 +-
.../flink/autoscaler/ScalingExecutorTest.java | 86 ++++++------
.../autoscaler/ScalingMetricCollectorTest.java | 32 +++--
.../autoscaler/ScalingMetricEvaluatorTest.java | 113 ++++++++++------
.../flink/autoscaler/TestScalingExecutor.java | 9 +-
.../flink/autoscaler/TestScalingExecutorTest.java | 30 +++--
.../flink/autoscaler/TestingMetricsCollector.java | 5 +-
.../autoscaler/alignment/AlignmentModeTest.java | 28 ++--
.../autoscaler/alignment/TestAlignmentMode.java | 2 +-
.../autoscaler/config/AutoScalerOptionsTest.java | 20 +++
.../autoscaler/metrics/TestCustomEvaluator.java | 4 +-
.../operator/autoscaler/AutoscalerFactory.java | 8 +-
.../kubernetes/operator/utils/AutoscalerUtils.java | 6 +-
39 files changed, 819 insertions(+), 503 deletions(-)
diff --git a/docs/content.zh/docs/operations/plugins.md b/docs/content.zh/docs/operations/plugins.md
index cf10afa8b..ae19bf2f4 100644
--- a/docs/content.zh/docs/operations/plugins.md
+++ b/docs/content.zh/docs/operations/plugins.md
@@ -215,12 +215,12 @@ The following steps demonstrate how to develop and use a custom metric evaluator
public Map<ScalingMetric, EvaluatedScalingMetric> evaluateVertexMetrics(
JobVertexID vertex,
Map<ScalingMetric, EvaluatedScalingMetric> evaluatedMetrics,
- Context context) {
+ Context<?> context) {
Map<ScalingMetric, EvaluatedScalingMetric> overrides = new HashMap<>();
// Example: override the target data rate for source vertices based
// on an evaluator-specific option read from the merged configuration.
- double target = context.getConfig().getDouble("target-data-rate", 0.0);
- if (target > 0 && context.getTopology().isSource(vertex)) {
+ double target = context.getConfiguration().getDouble("target-data-rate", 0.0);
+ if (target > 0 && context.getJobTopology().isSource(vertex)) {
overrides.put(
ScalingMetric.TARGET_DATA_RATE,
EvaluatedScalingMetric.avg(target));
@@ -230,7 +230,7 @@ The following steps demonstrate how to develop and use a custom metric evaluator
}
```
- The `Context` object exposes an un-modifiable view of the configuration, the metrics history, previously evaluated vertex metrics (evaluation happens topologically), the job topology, backlog status, and max restart time. The configuration is the job configuration with the evaluator-specific options (prefix stripped) merged on top, so an evaluator-specific key overrides the job-level value of the same key.
+ The `Context` extends `JobAutoScalerContext`, exposing the metric history, job topology and max restart time through the inherited accessors, plus the previously evaluated vertex metrics (evaluation happens topologically) and backlog status. Its `getConfiguration()` returns the job configuration with the evaluator-specific options (prefix stripped) merged on top, so an evaluator-specific key overrides the job-level value of the same key.
2. Create the service definition file `org.apache.flink.autoscaler.metrics.ScalingMetricsEvaluatorPlugin` in `META-INF/services` with the fully-qualified class name of your implementation:
```text
@@ -239,7 +239,7 @@ The following steps demonstrate how to develop and use a custom metric evaluator
3. Use the Maven tool to package the project and generate the custom metric evaluator JAR.
-4. Select the custom metric evaluator via configuration. The evaluator whose implementation class FQN matches the configured `job.autoscaler.metrics.custom-evaluator.<name>.class` value will be invoked, and any other `job.autoscaler.metrics.custom-evaluator.<name>.*` entries are merged on top of the job configuration with the `job.autoscaler.metrics.custom-evaluator.<instance>.` prefix stripped, and exposed via `Context.getConfig()`, taking precedence over the job-level value of the same [...]
+4. Select the custom metric evaluator via configuration. The evaluator whose implementation class FQN matches the configured `job.autoscaler.metrics.custom-evaluator.<name>.class` value will be invoked, and any other `job.autoscaler.metrics.custom-evaluator.<name>.*` entries are merged on top of the job configuration with the `job.autoscaler.metrics.custom-evaluator.<instance>.` prefix stripped, and exposed via `Context.getConfiguration()`, taking precedence over the job-level value of t [...]
```yaml
job.autoscaler.metrics.custom-evaluators: my-evaluator
job.autoscaler.metrics.custom-evaluator.my-evaluator.class: org.apache.flink.autoscaler.custom.CustomEvaluator
@@ -295,18 +295,18 @@ The following steps demonstrate how to develop and use a custom alignment mode.
/** Apply to every vertex, not only the source and keyBy vertices the built-ins handle. */
@Override
- public boolean isApplicable(Context ctx) {
+ public boolean isApplicable(Context<?> ctx) {
return true;
}
@Override
- public int align(Context ctx) {
+ public int alignParallelism(Context<?> ctx) {
// Number of key groups, or source partitions for a partitioned source.
int n = ctx.getNumSourcePartitions() > 0
? ctx.getNumSourcePartitions()
: ctx.getMaxParallelism();
// Optional mode-specific parameter, read from the per-mode configuration.
- int minParallelism = ctx.getModeConfiguration().getInteger("min-parallelism", 1);
+ int minParallelism = ctx.getConfiguration().getInteger("min-parallelism", 1);
for (int p = ctx.getNewParallelism(); p >= minParallelism; p--) {
if (n % p == 0) {
return p;
@@ -316,7 +316,7 @@ The following steps demonstrate how to develop and use a custom alignment mode.
}
}
```
- The `Context` exposes the current and computed target parallelism, the number of key groups or source partitions, the input ship strategies, the `JobAutoScalerContext`, the per-vertex evaluated metrics, the job topology, and the prefix-stripped per-mode `Configuration` (`getModeConfiguration()`). The autoscaler calls `align` only when `isApplicable(Context)` returns true. That method defaults to keyBy (hash) vertices and to partitioned sources that report a partition count (Kafka and [...]
+ The `Context` extends `JobAutoScalerContext`, adding the per-vertex alignment inputs: the current and computed target parallelism, the number of key groups or source partitions, the input ship strategies, and the vertex's evaluated metrics. The job topology is available through the inherited `getJobTopology()`, and `getConfiguration()` returns the job configuration with this mode's prefix-stripped per-mode options merged on top. The autoscaler calls `alignParallelism` only when `isApp [...]
2. Create the service definition file `org.apache.flink.autoscaler.alignment.ParallelismAlignmentMode` in `META-INF/services`:
```text
@@ -325,7 +325,7 @@ The following steps demonstrate how to develop and use a custom alignment mode.
3. Use the Maven tool to package the project and generate the custom alignment mode JAR.
-4. Select the custom mode by name and point it at your implementation class. Any other `job.autoscaler.scaling.parallelism-alignment.mode.<name>.*` entries are passed to the mode (prefix-stripped) through `Context#getModeConfiguration()`:
+4. Select the custom mode by name and point it at your implementation class. Any other `job.autoscaler.scaling.parallelism-alignment.mode.<name>.*` entries are merged (prefix-stripped) on top of the job configuration and exposed to the mode through `Context#getConfiguration()`:
```yaml
job.autoscaler.scaling.parallelism-alignment.mode: custom-mode
job.autoscaler.scaling.parallelism-alignment.mode.custom-mode.class: org.apache.flink.autoscaler.alignment.CustomAlignmentMode
@@ -378,7 +378,6 @@ The following steps demonstrate how to develop and use a custom scaling executor
```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;
@@ -387,8 +386,7 @@ The following steps demonstrate how to develop and use a custom scaling executor
import java.util.Map;
/** Custom implementation of {@link ScalingExecutorPlugin} that caps parallelism. */
- public class CustomScalingExecutor<KEY, CTX extends JobAutoScalerContext<KEY>>
- implements ScalingExecutorPlugin<KEY, CTX> {
+ public class CustomScalingExecutor<KEY> implements ScalingExecutorPlugin<KEY> {
@Override
public int priority() {
@@ -398,7 +396,7 @@ The following steps demonstrate how to develop and use a custom scaling executor
@Override
public Map<JobVertexID, ScalingSummary> apply(
- Context<KEY, CTX> context, Map<JobVertexID, ScalingSummary> scalingSummaries) {
+ Context<KEY> 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);
@@ -415,7 +413,7 @@ The following steps demonstrate how to develop and use a custom scaling executor
}
```
- 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`.
+ The `Context` extends `JobAutoScalerContext` for the current job, exposing the evaluated scaling metrics and job topology through the inherited accessors. Its `getConfiguration()` returns the job configuration with this plugin instance's prefix-stripped options merged on top.
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
diff --git a/docs/content/docs/operations/plugins.md b/docs/content/docs/operations/plugins.md
index 4c2f4a125..1ddcf4a79 100644
--- a/docs/content/docs/operations/plugins.md
+++ b/docs/content/docs/operations/plugins.md
@@ -215,12 +215,12 @@ The following steps demonstrate how to develop and use a custom metric evaluator
public Map<ScalingMetric, EvaluatedScalingMetric> evaluateVertexMetrics(
JobVertexID vertex,
Map<ScalingMetric, EvaluatedScalingMetric> evaluatedMetrics,
- Context context) {
+ Context<?> context) {
Map<ScalingMetric, EvaluatedScalingMetric> overrides = new HashMap<>();
// Example: override the target data rate for source vertices based
// on an evaluator-specific option read from the merged configuration.
- double target = context.getConfig().getDouble("target-data-rate", 0.0);
- if (target > 0 && context.getTopology().isSource(vertex)) {
+ double target = context.getConfiguration().getDouble("target-data-rate", 0.0);
+ if (target > 0 && context.getJobTopology().isSource(vertex)) {
overrides.put(
ScalingMetric.TARGET_DATA_RATE,
EvaluatedScalingMetric.avg(target));
@@ -230,7 +230,7 @@ The following steps demonstrate how to develop and use a custom metric evaluator
}
```
- The `Context` object exposes an un-modifiable view of the configuration, the metrics history, previously evaluated vertex metrics (evaluation happens topologically), the job topology, backlog status, and max restart time. The configuration is the job configuration with the evaluator-specific options (prefix stripped) merged on top, so an evaluator-specific key overrides the job-level value of the same key.
+ The `Context` extends `JobAutoScalerContext`, exposing the metric history, job topology and max restart time through the inherited accessors, plus the previously evaluated vertex metrics (evaluation happens topologically) and backlog status. Its `getConfiguration()` returns the job configuration with the evaluator-specific options (prefix stripped) merged on top, so an evaluator-specific key overrides the job-level value of the same key.
2. Create the service definition file `org.apache.flink.autoscaler.metrics.ScalingMetricsEvaluatorPlugin` in `META-INF/services` with the fully-qualified class name of your implementation:
```text
@@ -239,7 +239,7 @@ The following steps demonstrate how to develop and use a custom metric evaluator
3. Use the Maven tool to package the project and generate the custom metric evaluator JAR.
-4. Select the custom metric evaluator via configuration. The evaluator whose implementation class FQN matches the configured `job.autoscaler.metrics.custom-evaluator.<name>.class` value will be invoked, and any other `job.autoscaler.metrics.custom-evaluator.<name>.*` entries are merged on top of the job configuration with the `job.autoscaler.metrics.custom-evaluator.<instance>.` prefix stripped, and exposed via `Context.getConfig()`, taking precedence over the job-level value of the same [...]
+4. Select the custom metric evaluator via configuration. The evaluator whose implementation class FQN matches the configured `job.autoscaler.metrics.custom-evaluator.<name>.class` value will be invoked, and any other `job.autoscaler.metrics.custom-evaluator.<name>.*` entries are merged on top of the job configuration with the `job.autoscaler.metrics.custom-evaluator.<instance>.` prefix stripped, and exposed via `Context.getConfiguration()`, taking precedence over the job-level value of t [...]
```yaml
job.autoscaler.metrics.custom-evaluators: my-evaluator
job.autoscaler.metrics.custom-evaluator.my-evaluator.class: org.apache.flink.autoscaler.custom.CustomEvaluator
@@ -295,18 +295,18 @@ The following steps demonstrate how to develop and use a custom alignment mode.
/** Apply to every vertex, not only the source and keyBy vertices the built-ins handle. */
@Override
- public boolean isApplicable(Context ctx) {
+ public boolean isApplicable(Context<?> ctx) {
return true;
}
@Override
- public int align(Context ctx) {
+ public int alignParallelism(Context<?> ctx) {
// Number of key groups, or source partitions for a partitioned source.
int n = ctx.getNumSourcePartitions() > 0
? ctx.getNumSourcePartitions()
: ctx.getMaxParallelism();
// Optional mode-specific parameter, read from the per-mode configuration.
- int minParallelism = ctx.getModeConfiguration().getInteger("min-parallelism", 1);
+ int minParallelism = ctx.getConfiguration().getInteger("min-parallelism", 1);
for (int p = ctx.getNewParallelism(); p >= minParallelism; p--) {
if (n % p == 0) {
return p;
@@ -316,7 +316,7 @@ The following steps demonstrate how to develop and use a custom alignment mode.
}
}
```
- The `Context` exposes the current and computed target parallelism, the number of key groups or source partitions, the input ship strategies, the `JobAutoScalerContext`, the per-vertex evaluated metrics, the job topology, and the prefix-stripped per-mode `Configuration` (`getModeConfiguration()`). The autoscaler calls `align` only when `isApplicable(Context)` returns true. That method defaults to keyBy (hash) vertices and to partitioned sources that report a partition count (Kafka and [...]
+ The `Context` extends `JobAutoScalerContext`, adding the per-vertex alignment inputs: the current and computed target parallelism, the number of key groups or source partitions, the input ship strategies, and the vertex's evaluated metrics. The job topology is available through the inherited `getJobTopology()`, and `getConfiguration()` returns the job configuration with this mode's prefix-stripped per-mode options merged on top. The autoscaler calls `alignParallelism` only when `isApp [...]
2. Create the service definition file `org.apache.flink.autoscaler.alignment.ParallelismAlignmentMode` in `META-INF/services`:
```text
@@ -325,7 +325,7 @@ The following steps demonstrate how to develop and use a custom alignment mode.
3. Use the Maven tool to package the project and generate the custom alignment mode JAR.
-4. Select the custom mode by name and point it at your implementation class. Any other `job.autoscaler.scaling.parallelism-alignment.mode.<name>.*` entries are passed to the mode (prefix-stripped) through `Context#getModeConfiguration()`:
+4. Select the custom mode by name and point it at your implementation class. Any other `job.autoscaler.scaling.parallelism-alignment.mode.<name>.*` entries are merged (prefix-stripped) on top of the job configuration and exposed to the mode through `Context#getConfiguration()`:
```yaml
job.autoscaler.scaling.parallelism-alignment.mode: custom-mode
job.autoscaler.scaling.parallelism-alignment.mode.custom-mode.class: org.apache.flink.autoscaler.alignment.CustomAlignmentMode
@@ -378,7 +378,6 @@ The following steps demonstrate how to develop and use a custom scaling executor
```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;
@@ -387,8 +386,7 @@ The following steps demonstrate how to develop and use a custom scaling executor
import java.util.Map;
/** Custom implementation of {@link ScalingExecutorPlugin} that caps parallelism. */
- public class CustomScalingExecutor<KEY, CTX extends JobAutoScalerContext<KEY>>
- implements ScalingExecutorPlugin<KEY, CTX> {
+ public class CustomScalingExecutor<KEY> implements ScalingExecutorPlugin<KEY> {
@Override
public int priority() {
@@ -398,7 +396,7 @@ The following steps demonstrate how to develop and use a custom scaling executor
@Override
public Map<JobVertexID, ScalingSummary> apply(
- Context<KEY, CTX> context, Map<JobVertexID, ScalingSummary> scalingSummaries) {
+ Context<KEY> 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);
@@ -415,7 +413,7 @@ The following steps demonstrate how to develop and use a custom scaling executor
}
```
- 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`.
+ The `Context` extends `JobAutoScalerContext` for the current job, exposing the evaluated scaling metrics and job topology through the inherited accessors. Its `getConfiguration()` returns the job configuration with this plugin instance's prefix-stripped options merged on top.
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
diff --git a/docs/layouts/shortcodes/generated/auto_scaler_configuration.html b/docs/layouts/shortcodes/generated/auto_scaler_configuration.html
index 1c815cafa..db32d3793 100644
--- a/docs/layouts/shortcodes/generated/auto_scaler_configuration.html
+++ b/docs/layouts/shortcodes/generated/auto_scaler_configuration.html
@@ -252,7 +252,7 @@
<td><h5>job.autoscaler.scaling.parallelism-alignment.mode.<name>.<parameter></h5></td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
- <td>An arbitrary parameter passed to the custom alignment mode named <name>. All such parameters are made available to the mode (with the prefix stripped) through Context.getModeConfiguration().</td>
+ <td>An arbitrary parameter passed to the custom alignment mode named <name>. All such parameters are made available to the mode (with the prefix stripped) through Context.getConfiguration().</td>
</tr>
<tr>
<td><h5>job.autoscaler.scaling.parallelism-alignment.mode.<name>.class</h5></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 cb6d2c5f3..368454f47 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
@@ -96,14 +96,14 @@ public class StandaloneAutoscalerEntrypoint {
AutoScalerStateStore<KEY, Context> stateStore) {
Collection<ScalingMetricsEvaluatorPlugin> customEvaluators =
AutoscalerUtils.discoverCustomEvaluators();
- Collection<ScalingExecutorPlugin<KEY, Context>> customExecutors =
+ Collection<ScalingExecutorPlugin<KEY>> customExecutors =
AutoscalerUtils.discoverCustomScalingExecutors();
Collection<ParallelismAlignmentMode> customAlignmentModes =
AutoscalerUtils.discoverCustomAlignmentModes();
return new JobAutoScalerImpl<>(
- new RestApiMetricsCollector<>(),
- new ScalingMetricEvaluator(customEvaluators),
+ new RestApiMetricsCollector<>(stateStore),
+ new ScalingMetricEvaluator<>(customEvaluators),
new ScalingExecutor<>(
eventHandler, stateStore, null, customExecutors, customAlignmentModes),
eventHandler,
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 8d0df6b5f..2bd609901 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
@@ -65,16 +65,15 @@ public class AutoscalerUtils {
*/
@SuppressWarnings("unchecked")
public static <KEY, Context extends JobAutoScalerContext<KEY>>
- Collection<ScalingExecutorPlugin<KEY, Context>> discoverCustomScalingExecutors() {
- List<ScalingExecutorPlugin<KEY, Context>> customScalingExecutors = new ArrayList<>();
+ Collection<ScalingExecutorPlugin<KEY>> discoverCustomScalingExecutors() {
+ List<ScalingExecutorPlugin<KEY>> 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);
+ customScalingExecutors.add((ScalingExecutorPlugin<KEY>) 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 f64b045a9..b31b5cd50 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,7 +17,6 @@
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.alignment.ParallelismAlignmentMode;
@@ -47,7 +46,7 @@ class AutoscalerUtilsTest {
@Test
void testDiscoverCustomScalingExecutors() {
- Collection<ScalingExecutorPlugin<Object, JobAutoScalerContext<Object>>> scalingExecutors =
+ Collection<ScalingExecutorPlugin<Object>> scalingExecutors =
AutoscalerUtils.discoverCustomScalingExecutors();
assertThat(scalingExecutors)
diff --git a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobAutoScalerContext.java b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobAutoScalerContext.java
index 9c5e614c8..1bb81b379 100644
--- a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobAutoScalerContext.java
+++ b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobAutoScalerContext.java
@@ -20,21 +20,34 @@ package org.apache.flink.autoscaler;
import org.apache.flink.annotation.Experimental;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.autoscaler.config.AutoScalerOptions;
+import org.apache.flink.autoscaler.metrics.AutoscalerFlinkMetrics;
+import org.apache.flink.autoscaler.metrics.CollectedMetricHistory;
+import org.apache.flink.autoscaler.metrics.CollectedMetrics;
+import org.apache.flink.autoscaler.metrics.EvaluatedMetrics;
+import org.apache.flink.autoscaler.topology.JobTopology;
import org.apache.flink.client.program.rest.RestClusterClient;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.util.function.SupplierWithException;
+import lombok.AccessLevel;
import lombok.Builder;
import lombok.Getter;
-import lombok.RequiredArgsConstructor;
+import lombok.Setter;
import lombok.ToString;
import javax.annotation.Nullable;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Collections;
+import java.util.Map;
import java.util.Optional;
+import java.util.SortedMap;
/**
* The job autoscaler context, it includes all details related to the current job.
@@ -42,9 +55,7 @@ import java.util.Optional;
* @param <KEY> The job key.
*/
@Experimental
-@RequiredArgsConstructor
@ToString
-@Builder(toBuilder = true)
public class JobAutoScalerContext<KEY> {
/** The identifier of each flink job. */
@@ -58,14 +69,63 @@ public class JobAutoScalerContext<KEY> {
/**
* The configuration based on the latest user-provided spec. This is not the already deployed /
* observed configuration.
+ *
+ * <p>For a plugin context this is the effective configuration: the job configuration with that
+ * plugin's prefix-stripped per-plugin overrides merged on top.
*/
@Getter private final Configuration configuration;
+ /** The metric group for the current job, used to register autoscaler and plugin metrics. */
@Getter private final MetricGroup metricGroup;
+ /** Supplier of a Flink REST client for the current job, invoked lazily on first use. */
@ToString.Exclude
private final SupplierWithException<RestClusterClient<String>, Exception> restClientSupplier;
+ /**
+ * Internal scratch space populated by the autoscaler pipeline within a single scaling cycle. It
+ * is lazily initialized on first access and excluded from {@code toString}, and is not part of
+ * the externally-provided context (callers should not set it).
+ */
+ @ToString.Exclude private ScalingCycleState scalingCycleState;
+
+ @Builder(toBuilder = true)
+ public JobAutoScalerContext(
+ KEY jobKey,
+ @Nullable JobID jobID,
+ @Nullable JobStatus jobStatus,
+ Configuration configuration,
+ MetricGroup metricGroup,
+ SupplierWithException<RestClusterClient<String>, Exception> restClientSupplier) {
+ this.jobKey = jobKey;
+ this.jobID = jobID;
+ this.jobStatus = jobStatus;
+ this.configuration = configuration;
+ this.metricGroup = metricGroup;
+ this.restClientSupplier = restClientSupplier;
+ }
+
+ /**
+ * Copy constructor used by plugin context subclasses. It copies all fields from the given
+ * context, overlaying the plugin's prefix-stripped {@code overrides} on top of that context's
+ * configuration (via {@link AutoScalerOptions#overlayConfiguration}), so a plugin context
+ * exposes its effective per-plugin configuration through {@code getConfiguration()} without
+ * each caller having to merge it. It also shares the {@link ScalingCycleState}, so the derived
+ * context exposes the same per-cycle data (collected / evaluated metrics, topology, restart
+ * time) through the inherited accessors rather than duplicating it.
+ */
+ protected JobAutoScalerContext(
+ JobAutoScalerContext<KEY> autoScalerContext, Configuration overrides) {
+ this.jobKey = autoScalerContext.jobKey;
+ this.jobID = autoScalerContext.jobID;
+ this.jobStatus = autoScalerContext.jobStatus;
+ this.configuration =
+ AutoScalerOptions.overlayConfiguration(autoScalerContext.configuration, overrides);
+ this.metricGroup = autoScalerContext.metricGroup;
+ this.restClientSupplier = autoScalerContext.restClientSupplier;
+ this.scalingCycleState = autoScalerContext.getScalingCycleState();
+ }
+
/** Retrieve the currently configured TaskManager CPU. */
public Optional<Double> getTaskManagerCpu() {
// Not supported by default
@@ -80,4 +140,85 @@ public class JobAutoScalerContext<KEY> {
public RestClusterClient<String> getRestClusterClient() throws Exception {
return restClientSupplier.get();
}
+
+ /**
+ * Returns the mutable per-cycle {@link ScalingCycleState} for this context, lazily creating it
+ * on first access. The metric collector, evaluator and scaling executor read from and write to
+ * it so they can rely solely on the context threaded through them.
+ */
+ public ScalingCycleState getScalingCycleState() {
+ if (scalingCycleState == null) {
+ scalingCycleState = new ScalingCycleState();
+ }
+ return scalingCycleState;
+ }
+
+ /**
+ * The job topology collected for the current cycle, derived from the {@link ScalingCycleState}.
+ * Only meaningful once metrics have been collected this cycle. Intended for use by plugin
+ * implementations.
+ */
+ public JobTopology getJobTopology() {
+ return getScalingCycleState().getCollectedMetrics().getJobTopology();
+ }
+
+ /**
+ * The metrics evaluated for the current cycle, derived from the {@link ScalingCycleState}. Only
+ * meaningful once evaluation has run this cycle. Intended for use by plugin implementations.
+ */
+ public EvaluatedMetrics getEvaluatedMetrics() {
+ return getScalingCycleState().getEvaluatedMetrics();
+ }
+
+ /**
+ * An unmodifiable view of the collected metric history for the current cycle, derived from the
+ * {@link ScalingCycleState}. Intended for use by plugin implementations.
+ */
+ public SortedMap<Instant, CollectedMetrics> getMetricsHistory() {
+ return Collections.unmodifiableSortedMap(
+ getScalingCycleState().getCollectedMetrics().getMetricHistory());
+ }
+
+ /**
+ * The restart time for the current cycle, derived from the {@link ScalingCycleState}. Intended
+ * for use by plugin implementations.
+ */
+ public Duration getRestartTime() {
+ return getScalingCycleState().getRestartTime();
+ }
+
+ /**
+ * Mutable, per-cycle working state of the autoscaler pipeline. A fresh instance is lazily
+ * created for every {@link JobAutoScalerContext} and progressively populated as a scaling cycle
+ * advances through metric collection, evaluation and scaling execution.
+ *
+ * <p>Threading this state through the context is what lets {@link ScalingMetricCollector},
+ * {@link ScalingMetricEvaluator} and {@link ScalingExecutor} rely solely on the context passed
+ * to them rather than on a long parameter list.
+ */
+ @Getter
+ @Setter(AccessLevel.PACKAGE)
+ public static class ScalingCycleState {
+
+ /** The instant at which the current scaling cycle started, shared by all stages. */
+ @Nullable private Instant now;
+
+ /** The Flink metrics sink for the current job, used to report scaling metrics. */
+ @Nullable private AutoscalerFlinkMetrics autoscalerMetrics;
+
+ /** The metrics collected from the Flink cluster during the current cycle. */
+ @Nullable private CollectedMetricHistory collectedMetrics;
+
+ /** The scaling tracking loaded (and possibly updated) during the current cycle. */
+ @Nullable private ScalingTracking scalingTracking;
+
+ /** The trimmed scaling history loaded during the current cycle. */
+ @Nullable private Map<JobVertexID, SortedMap<Instant, ScalingSummary>> scalingHistory;
+
+ /** The restart time used by both metric evaluation and scaling execution. */
+ @Nullable private Duration restartTime;
+
+ /** The metrics evaluated from {@link #collectedMetrics}. */
+ @Nullable private EvaluatedMetrics evaluatedMetrics;
+ }
}
diff --git a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobAutoScalerImpl.java b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobAutoScalerImpl.java
index b99f4da36..d516a82a5 100644
--- a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobAutoScalerImpl.java
+++ b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobAutoScalerImpl.java
@@ -23,7 +23,6 @@ import org.apache.flink.autoscaler.config.AutoScalerOptions;
import org.apache.flink.autoscaler.event.AutoScalerEventHandler;
import org.apache.flink.autoscaler.exceptions.NotReadyException;
import org.apache.flink.autoscaler.metrics.AutoscalerFlinkMetrics;
-import org.apache.flink.autoscaler.metrics.EvaluatedMetrics;
import org.apache.flink.autoscaler.realizer.ScalingRealizer;
import org.apache.flink.autoscaler.state.AutoScalerStateStore;
import org.apache.flink.autoscaler.tuning.ConfigChanges;
@@ -39,10 +38,6 @@ import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import static org.apache.flink.autoscaler.config.AutoScalerOptions.AUTOSCALER_ENABLED;
-import static org.apache.flink.autoscaler.metrics.AutoscalerFlinkMetrics.initRecommendedParallelism;
-import static org.apache.flink.autoscaler.metrics.AutoscalerFlinkMetrics.resetRecommendedParallelism;
-import static org.apache.flink.autoscaler.metrics.ScalingHistoryUtils.getTrimmedScalingHistory;
-import static org.apache.flink.autoscaler.metrics.ScalingHistoryUtils.getTrimmedScalingTracking;
/** The default implementation of {@link JobAutoScaler}. */
public class JobAutoScalerImpl<KEY, Context extends JobAutoScalerContext<KEY>>
@@ -53,7 +48,7 @@ public class JobAutoScalerImpl<KEY, Context extends JobAutoScalerContext<KEY>>
@VisibleForTesting protected static final String AUTOSCALER_ERROR = "AutoscalerError";
private final ScalingMetricCollector<KEY, Context> metricsCollector;
- private final ScalingMetricEvaluator evaluator;
+ @VisibleForTesting final ScalingMetricEvaluator<KEY, Context> metricsEvaluator;
private final ScalingExecutor<KEY, Context> scalingExecutor;
private final AutoScalerEventHandler<KEY, Context> eventHandler;
private final ScalingRealizer<KEY, Context> scalingRealizer;
@@ -61,21 +56,18 @@ public class JobAutoScalerImpl<KEY, Context extends JobAutoScalerContext<KEY>>
private Clock clock = Clock.systemDefaultZone();
- @VisibleForTesting
- final Map<KEY, EvaluatedMetrics> lastEvaluatedMetrics = new ConcurrentHashMap<>();
-
@VisibleForTesting
final Map<KEY, AutoscalerFlinkMetrics> flinkMetrics = new ConcurrentHashMap<>();
public JobAutoScalerImpl(
ScalingMetricCollector<KEY, Context> metricsCollector,
- ScalingMetricEvaluator evaluator,
+ ScalingMetricEvaluator<KEY, Context> metricsEvaluator,
ScalingExecutor<KEY, Context> scalingExecutor,
AutoScalerEventHandler<KEY, Context> eventHandler,
ScalingRealizer<KEY, Context> scalingRealizer,
AutoScalerStateStore<KEY, Context> stateStore) {
this.metricsCollector = metricsCollector;
- this.evaluator = evaluator;
+ this.metricsEvaluator = metricsEvaluator;
this.scalingExecutor = scalingExecutor;
this.eventHandler = eventHandler;
this.scalingRealizer = scalingRealizer;
@@ -87,7 +79,7 @@ public class JobAutoScalerImpl<KEY, Context extends JobAutoScalerContext<KEY>>
var autoscalerMetrics = getOrInitAutoscalerFlinkMetrics(ctx);
try {
- if (!ctx.getConfiguration().getBoolean(AUTOSCALER_ENABLED)) {
+ if (!ctx.getConfiguration().get(AUTOSCALER_ENABLED)) {
LOG.debug("Autoscaler is disabled");
stateStore.clearAll(ctx);
stateStore.flush(ctx);
@@ -96,7 +88,7 @@ public class JobAutoScalerImpl<KEY, Context extends JobAutoScalerContext<KEY>>
if (ctx.getJobStatus() != JobStatus.RUNNING) {
LOG.debug("Autoscaler is waiting for stable, running state");
- lastEvaluatedMetrics.remove(ctx.getJobKey());
+ metricsEvaluator.cleanup(ctx.getJobKey());
return;
}
@@ -121,7 +113,7 @@ public class JobAutoScalerImpl<KEY, Context extends JobAutoScalerContext<KEY>>
public void cleanup(Context ctx) {
LOG.info("Cleaning up autoscaling meta data");
metricsCollector.cleanup(ctx.getJobKey());
- lastEvaluatedMetrics.remove(ctx.getJobKey());
+ metricsEvaluator.cleanup(ctx.getJobKey());
flinkMetrics.remove(ctx.getJobKey());
try {
stateStore.clearAll(ctx);
@@ -179,63 +171,17 @@ public class JobAutoScalerImpl<KEY, Context extends JobAutoScalerContext<KEY>>
private void runScalingLogic(Context ctx, AutoscalerFlinkMetrics autoscalerMetrics)
throws Exception {
+ var cycleState = ctx.getScalingCycleState();
+ cycleState.setNow(clock.instant());
+ cycleState.setAutoscalerMetrics(autoscalerMetrics);
- var collectedMetrics = metricsCollector.updateMetrics(ctx, stateStore);
- var jobTopology = collectedMetrics.getJobTopology();
-
- var now = clock.instant();
- var scalingTracking = getTrimmedScalingTracking(stateStore, ctx, now);
- var scalingHistory = getTrimmedScalingHistory(stateStore, ctx, now);
- // A scaling tracking without an end time gets created whenever a scaling decision is
- // applied. Here, we record the end time for it (runScalingLogic is only called when the job
- // transitions back into the RUNNING state).
- if (scalingTracking.recordRestartDurationIfTrackedAndParallelismMatches(
- collectedMetrics.getJobRunningTs(), jobTopology, scalingHistory)) {
- stateStore.storeScalingTracking(ctx, scalingTracking);
- }
-
- // We require at least 2 metric collections for evaluation as required by rate computations
- // from accumulated metrics
- if (collectedMetrics.getMetricHistory().size() < 2) {
+ if (!metricsCollector.collect(ctx)) {
return;
}
- LOG.debug("Collected metrics: {}", collectedMetrics);
-
- // Scaling tracking data contains previous restart times that are taken into account
- var restartTime = scalingTracking.getMaxRestartTimeOrDefault(ctx.getConfiguration());
-
- var evaluatedMetrics =
- evaluator.evaluate(ctx.getConfiguration(), collectedMetrics, restartTime);
- LOG.debug("Evaluated metrics: {}", evaluatedMetrics);
- lastEvaluatedMetrics.put(ctx.getJobKey(), evaluatedMetrics);
-
- initRecommendedParallelism(evaluatedMetrics.getVertexMetrics());
- autoscalerMetrics.registerScalingMetrics(
- jobTopology.getVerticesInTopologicalOrder(),
- () -> lastEvaluatedMetrics.get(ctx.getJobKey()));
-
- if (!collectedMetrics.isFullyCollected()) {
- // We have done an upfront evaluation, but we are not ready for scaling.
- resetRecommendedParallelism(evaluatedMetrics.getVertexMetrics());
+ if (!metricsEvaluator.evaluate(ctx)) {
return;
}
-
- var delayedScaleDown = stateStore.getDelayedScaleDown(ctx);
- var parallelismChanged =
- scalingExecutor.scaleResource(
- ctx,
- evaluatedMetrics,
- scalingHistory,
- scalingTracking,
- now,
- jobTopology,
- delayedScaleDown);
-
- if (delayedScaleDown.isUpdated()) {
- stateStore.storeDelayedScaleDown(ctx, delayedScaleDown);
- }
-
- if (parallelismChanged) {
+ if (scalingExecutor.execute(ctx)) {
autoscalerMetrics.incrementScaling();
} else {
autoscalerMetrics.incrementBalanced();
diff --git a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobVertexScaler.java b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobVertexScaler.java
index 1aada42c8..6797e1e86 100644
--- a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobVertexScaler.java
+++ b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/JobVertexScaler.java
@@ -582,7 +582,6 @@ public class JobVertexScaler<KEY, Context extends JobAutoScalerContext<KEY>> {
parallelismUpperLimit,
inputShipStrategies,
evaluatedMetrics,
- jobTopology,
autoScalerEventHandler,
context);
}
diff --git a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/RestApiMetricsCollector.java b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/RestApiMetricsCollector.java
index 7ebedb76b..daeed3e5b 100644
--- a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/RestApiMetricsCollector.java
+++ b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/RestApiMetricsCollector.java
@@ -18,6 +18,7 @@
package org.apache.flink.autoscaler;
import org.apache.flink.autoscaler.metrics.FlinkMetric;
+import org.apache.flink.autoscaler.state.AutoScalerStateStore;
import org.apache.flink.client.program.rest.RestClusterClient;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
@@ -68,6 +69,10 @@ public class RestApiMetricsCollector<KEY, Context extends JobAutoScalerContext<K
.put("Status.JVM.GarbageCollector.All.TimeMsPerSecond", TOTAL_GC_TIME_PER_SEC)
.build();
+ public RestApiMetricsCollector(AutoScalerStateStore<KEY, Context> stateStore) {
+ super(stateStore);
+ }
+
@Override
protected Map<JobVertexID, Map<FlinkMetric, AggregatedMetric>> queryAllAggregatedMetrics(
Context ctx, Map<JobVertexID, Map<String, FlinkMetric>> filteredVertexMetricNames) {
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 f99eb3702..8843cef87 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
@@ -87,7 +87,7 @@ public class ScalingExecutor<KEY, Context extends JobAutoScalerContext<KEY>> {
private final AutoScalerEventHandler<KEY, Context> autoScalerEventHandler;
private final AutoScalerStateStore<KEY, Context> autoScalerStateStore;
private final ResourceCheck resourceCheck;
- private final Collection<ScalingExecutorPlugin<KEY, Context>> customExecutors;
+ private final Collection<ScalingExecutorPlugin<KEY>> customExecutors;
public ScalingExecutor(
AutoScalerEventHandler<KEY, Context> autoScalerEventHandler,
@@ -116,7 +116,7 @@ public class ScalingExecutor<KEY, Context extends JobAutoScalerContext<KEY>> {
AutoScalerEventHandler<KEY, Context> autoScalerEventHandler,
AutoScalerStateStore<KEY, Context> autoScalerStateStore,
@Nullable ResourceCheck resourceCheck,
- Collection<ScalingExecutorPlugin<KEY, Context>> customExecutors,
+ Collection<ScalingExecutorPlugin<KEY>> customExecutors,
Collection<ParallelismAlignmentMode> customAlignmentModes) {
this.jobVertexScaler = new JobVertexScaler<>(autoScalerEventHandler, customAlignmentModes);
this.autoScalerEventHandler = autoScalerEventHandler;
@@ -128,7 +128,36 @@ public class ScalingExecutor<KEY, Context extends JobAutoScalerContext<KEY>> {
: Collections.unmodifiableCollection(customExecutors);
}
- public boolean scaleResource(
+ /**
+ * Computes and applies the scaling decision for the current cycle based on the scaling cycle
+ * state of the given context. The delayed scale down state is loaded from and persisted to the
+ * {@link AutoScalerStateStore} owned by this executor.
+ *
+ * @return {@code true} if a parallelism change was applied, {@code false} otherwise.
+ */
+ public boolean execute(Context context) throws Exception {
+ var cycleState = context.getScalingCycleState();
+ var delayedScaleDown = autoScalerStateStore.getDelayedScaleDown(context);
+
+ var parallelismChanged =
+ scaleResource(
+ context,
+ cycleState.getEvaluatedMetrics(),
+ cycleState.getScalingHistory(),
+ cycleState.getScalingTracking(),
+ cycleState.getNow(),
+ cycleState.getCollectedMetrics().getJobTopology(),
+ delayedScaleDown);
+
+ if (delayedScaleDown.isUpdated()) {
+ autoScalerStateStore.storeDelayedScaleDown(context, delayedScaleDown);
+ }
+
+ return parallelismChanged;
+ }
+
+ @VisibleForTesting
+ boolean scaleResource(
Context context,
EvaluatedMetrics evaluatedMetrics,
Map<JobVertexID, SortedMap<Instant, ScalingSummary>> scalingHistory,
@@ -182,8 +211,6 @@ public class ScalingExecutor<KEY, Context extends JobAutoScalerContext<KEY>> {
applyCustomExecutors(
context,
memoryTuningEnabled ? configOverrides.newConfigWithOverrides(conf) : conf,
- evaluatedMetrics,
- jobTopology,
scalingSummaries);
if (scalingSummaries == null || scalingSummaries.isEmpty()) {
return false;
@@ -523,23 +550,21 @@ public class ScalingExecutor<KEY, Context extends JobAutoScalerContext<KEY>> {
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 =
+ Collection<Map.Entry<String, ScalingExecutorPlugin<KEY>>> chain =
resolveCustomScalingExecutors(conf);
if (chain.isEmpty()) {
return scalingSummaries;
}
- for (Map.Entry<String, ScalingExecutorPlugin<KEY, Context>> entry : chain) {
+ for (Map.Entry<String, ScalingExecutorPlugin<KEY>> 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);
+ context,
+ AutoScalerOptions.customScalingExecutorConfiguration(
+ conf, instanceName));
var previousSummaries = copyScalingSummaries(scalingSummaries);
scalingSummaries = plugin.apply(pluginContext, scalingSummaries);
if (scalingSummaries == null || scalingSummaries.isEmpty()) {
@@ -580,13 +605,13 @@ public class ScalingExecutor<KEY, Context extends JobAutoScalerContext<KEY>> {
* event {@code messageKey}, hence the {@code Map.Entry} pairing.
*/
@VisibleForTesting
- Collection<Map.Entry<String, ScalingExecutorPlugin<KEY, Context>>>
- resolveCustomScalingExecutors(Configuration conf) {
+ Collection<Map.Entry<String, ScalingExecutorPlugin<KEY>>> 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<>();
+ List<Map.Entry<String, ScalingExecutorPlugin<KEY>>> resolved = new ArrayList<>();
Set<String> seenInstanceNames = new HashSet<>();
for (String instance : instances) {
if (!seenInstanceNames.add(instance)) {
@@ -607,8 +632,8 @@ public class ScalingExecutor<KEY, Context extends JobAutoScalerContext<KEY>> {
classKey);
continue;
}
- ScalingExecutorPlugin<KEY, Context> match = null;
- for (ScalingExecutorPlugin<KEY, Context> plugin : customExecutors) {
+ ScalingExecutorPlugin<KEY> match = null;
+ for (ScalingExecutorPlugin<KEY> plugin : customExecutors) {
if (plugin.getClass().getName().equals(configuredClassName)) {
match = plugin;
break;
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
index d5a471392..7f3d36211 100644
--- a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingExecutorPlugin.java
+++ b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingExecutorPlugin.java
@@ -17,13 +17,10 @@
package org.apache.flink.autoscaler;
-import org.apache.flink.autoscaler.metrics.EvaluatedMetrics;
-import org.apache.flink.autoscaler.topology.JobTopology;
+import org.apache.flink.annotation.Experimental;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.jobgraph.JobVertexID;
-import lombok.Value;
-
import java.util.Collections;
import java.util.Map;
@@ -49,9 +46,9 @@ import java.util.Map;
* 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>> {
+@Experimental
+public interface ScalingExecutorPlugin<KEY> {
/**
* Returns the priority of this plugin in the chain. Plugins with lower priority values are
@@ -80,41 +77,33 @@ public interface ScalingExecutorPlugin<KEY, CTX extends JobAutoScalerContext<KEY
* operation entirely.
* </ul>
*
- * @param context The plugin context wrapping the autoscaler context, configuration, evaluated
- * metrics, and job topology.
+ * @param context The plugin context, a {@link JobAutoScalerContext} extended with the
+ * per-plugin configuration and convenience access to the 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);
+ Context<KEY> 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.
+ * The scaling executor plugin context. It {@code extends} {@link JobAutoScalerContext}, sharing
+ * its {@link org.apache.flink.autoscaler.JobAutoScalerContext.ScalingCycleState} and inherited
+ * cycle accessors (such as {@link #getEvaluatedMetrics()} and {@link #getJobTopology()}). Its
+ * {@code getConfiguration()} returns the effective per-plugin configuration: the job
+ * configuration with this plugin's prefix-stripped {@code
+ * job.autoscaler.scaling.custom-executor.<name>.<parameter>} overrides (the {@code
+ * kubernetes.operator.} legacy fallback honored) merged on top, so plugin-specific keys take
+ * precedence.
*
* @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;
+ class Context<KEY> extends JobAutoScalerContext<KEY> {
- /** The job topology. */
- JobTopology jobTopology;
+ public Context(JobAutoScalerContext<KEY> autoScalerContext, Configuration overrides) {
+ super(autoScalerContext, overrides);
+ }
}
}
diff --git a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricCollector.java b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricCollector.java
index d538ab63d..9cdf4ace3 100644
--- a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricCollector.java
+++ b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricCollector.java
@@ -70,6 +70,8 @@ import java.util.regex.Pattern;
import java.util.stream.Collectors;
import java.util.stream.Stream;
+import static org.apache.flink.autoscaler.metrics.ScalingHistoryUtils.getTrimmedScalingHistory;
+import static org.apache.flink.autoscaler.metrics.ScalingHistoryUtils.getTrimmedScalingTracking;
import static org.apache.flink.autoscaler.metrics.ScalingHistoryUtils.updateVertexList;
import static org.apache.flink.autoscaler.utils.AutoScalerUtils.excludeVerticesFromScaling;
import static org.apache.flink.autoscaler.utils.DateTimeUtils.readable;
@@ -85,16 +87,31 @@ public abstract class ScalingMetricCollector<KEY, Context extends JobAutoScalerC
new ConcurrentHashMap<>();
protected final Map<KEY, Boolean> jobsWithGcMetrics = new ConcurrentHashMap<KEY, Boolean>();
+ private final AutoScalerStateStore<KEY, Context> stateStore;
+
private Clock clock = Clock.systemDefaultZone();
- public CollectedMetricHistory updateMetrics(
- Context ctx, AutoScalerStateStore<KEY, Context> stateStore) throws Exception {
+ protected ScalingMetricCollector(AutoScalerStateStore<KEY, Context> stateStore) {
+ this.stateStore = stateStore;
+ }
+
+ /**
+ * Collects the latest metrics for the job and records them in the scaling cycle state of the
+ * given context (collected metrics, scaling tracking and history, and the restart time).
+ *
+ * @return {@code true} if enough metric samples have been collected for evaluation, {@code
+ * false} if the caller should stop the current scaling cycle early.
+ */
+ public boolean collect(Context ctx) throws Exception {
+ var cycleState = ctx.getScalingCycleState();
+ if (cycleState.getNow() == null) {
+ cycleState.setNow(clock.instant());
+ }
+ var now = cycleState.getNow();
var jobKey = ctx.getJobKey();
var conf = ctx.getConfiguration();
- var now = clock.instant();
-
var metricHistory =
histories.computeIfAbsent(
jobKey,
@@ -118,7 +135,7 @@ public abstract class ScalingMetricCollector<KEY, Context extends JobAutoScalerC
cleanup(ctx.getJobKey());
metricHistory.clear();
}
- var topology = getJobTopology(ctx, stateStore, jobDetailsInfo);
+ var topology = getJobTopology(ctx, jobDetailsInfo);
var stableTime = jobRunningTs.plus(conf.get(AutoScalerOptions.STABILIZATION_INTERVAL));
final boolean isStabilizing = now.isBefore(stableTime);
@@ -171,7 +188,32 @@ public abstract class ScalingMetricCollector<KEY, Context extends JobAutoScalerC
metricHistory.size());
}
stateStore.storeCollectedMetrics(ctx, metricHistory);
- return collectedMetrics;
+ cycleState.setCollectedMetrics(collectedMetrics);
+
+ // Scaling tracking and history are loaded here so the rest of the pipeline (evaluation and
+ // execution) can read them from the cycle state.
+ var scalingTracking = getTrimmedScalingTracking(stateStore, ctx, now);
+ var scalingHistory = getTrimmedScalingHistory(stateStore, ctx, now);
+ cycleState.setScalingTracking(scalingTracking);
+ cycleState.setScalingHistory(scalingHistory);
+ // A scaling tracking without an end time gets created whenever a scaling decision is
+ // applied. Here, we record the end time for it (collect is only called when the job
+ // transitions back into the RUNNING state).
+ if (scalingTracking.recordRestartDurationIfTrackedAndParallelismMatches(
+ jobRunningTs, topology, scalingHistory)) {
+ stateStore.storeScalingTracking(ctx, scalingTracking);
+ }
+
+ // We require at least 2 metric collections for evaluation as required by rate computations
+ // from accumulated metrics.
+ if (metricHistory.size() < 2) {
+ return false;
+ }
+ LOG.debug("Collected metrics: {}", collectedMetrics);
+
+ // Scaling tracking data contains previous restart times that are taken into account.
+ cycleState.setRestartTime(scalingTracking.getMaxRestartTimeOrDefault(conf));
+ return true;
}
private int removeMetricsBefore(
@@ -209,10 +251,7 @@ public abstract class ScalingMetricCollector<KEY, Context extends JobAutoScalerC
return Instant.ofEpochMilli(runningTs);
}
- protected JobTopology getJobTopology(
- Context ctx,
- AutoScalerStateStore<KEY, Context> stateStore,
- JobDetailsInfo jobDetailsInfo)
+ protected JobTopology getJobTopology(Context ctx, JobDetailsInfo jobDetailsInfo)
throws Exception {
var t = getJobTopology(jobDetailsInfo);
diff --git a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricEvaluator.java b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricEvaluator.java
index 4a1083742..15a85b9cd 100644
--- a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricEvaluator.java
+++ b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricEvaluator.java
@@ -30,7 +30,6 @@ import org.apache.flink.autoscaler.metrics.ScalingMetricsEvaluatorPlugin;
import org.apache.flink.autoscaler.topology.JobTopology;
import org.apache.flink.autoscaler.utils.AutoScalerUtils;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.UnmodifiableConfiguration;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.slf4j.Logger;
@@ -48,6 +47,7 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.SortedMap;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
import static org.apache.flink.autoscaler.config.AutoScalerOptions.BACKLOG_PROCESSING_LAG_THRESHOLD;
@@ -56,6 +56,8 @@ import static org.apache.flink.autoscaler.config.AutoScalerOptions.TARGET_UTILIZ
import static org.apache.flink.autoscaler.config.AutoScalerOptions.UTILIZATION_MAX;
import static org.apache.flink.autoscaler.config.AutoScalerOptions.UTILIZATION_MIN;
import static org.apache.flink.autoscaler.config.AutoScalerOptions.UTILIZATION_TARGET;
+import static org.apache.flink.autoscaler.metrics.AutoscalerFlinkMetrics.initRecommendedParallelism;
+import static org.apache.flink.autoscaler.metrics.AutoscalerFlinkMetrics.resetRecommendedParallelism;
import static org.apache.flink.autoscaler.metrics.ScalingMetric.CATCH_UP_DATA_RATE;
import static org.apache.flink.autoscaler.metrics.ScalingMetric.GC_PRESSURE;
import static org.apache.flink.autoscaler.metrics.ScalingMetric.HEAP_MAX_USAGE_RATIO;
@@ -75,18 +77,75 @@ import static org.apache.flink.autoscaler.metrics.ScalingMetric.TARGET_DATA_RATE
import static org.apache.flink.autoscaler.metrics.ScalingMetric.TRUE_PROCESSING_RATE;
/** Job scaling evaluator for autoscaler. */
-public class ScalingMetricEvaluator {
+public class ScalingMetricEvaluator<KEY, Context extends JobAutoScalerContext<KEY>> {
private static final Logger LOG = LoggerFactory.getLogger(ScalingMetricEvaluator.class);
private final Collection<ScalingMetricsEvaluatorPlugin> customEvaluators;
+ /**
+ * The most recently evaluated metrics per job. Retained across scaling cycles so the scaling
+ * metric gauges (which are registered only once) always report the latest values.
+ */
+ @VisibleForTesting
+ final Map<KEY, EvaluatedMetrics> lastEvaluatedMetrics = new ConcurrentHashMap<>();
+
public ScalingMetricEvaluator(Collection<ScalingMetricsEvaluatorPlugin> customEvaluators) {
this.customEvaluators = customEvaluators;
}
- public EvaluatedMetrics evaluate(
- Configuration conf, CollectedMetricHistory collectedMetrics, Duration restartTime) {
+ /**
+ * The production evaluation entry point. Evaluates the metrics collected during the current
+ * cycle (running the configured custom evaluator plugin, if any), records the result in the
+ * scaling cycle state of the given context, and registers / updates the scaling metric gauges.
+ *
+ * @return {@code true} if the metrics are fully collected and the job is ready for scaling,
+ * {@code false} if the caller should stop the current scaling cycle early. An upfront
+ * evaluation is still performed and reported in the latter case.
+ */
+ public boolean evaluate(Context ctx) {
+ var cycleState = ctx.getScalingCycleState();
+ var collectedMetrics = cycleState.getCollectedMetrics();
+
+ var evaluatedMetrics =
+ computeEvaluatedMetrics(
+ ctx, ctx.getConfiguration(), collectedMetrics, cycleState.getRestartTime());
+ LOG.debug("Evaluated metrics: {}", evaluatedMetrics);
+ cycleState.setEvaluatedMetrics(evaluatedMetrics);
+ lastEvaluatedMetrics.put(ctx.getJobKey(), evaluatedMetrics);
+
+ initRecommendedParallelism(evaluatedMetrics.getVertexMetrics());
+ cycleState
+ .getAutoscalerMetrics()
+ .registerScalingMetrics(
+ collectedMetrics.getJobTopology().getVerticesInTopologicalOrder(),
+ () -> lastEvaluatedMetrics.get(ctx.getJobKey()));
+
+ if (!collectedMetrics.isFullyCollected()) {
+ // We have done an upfront evaluation, but we are not ready for scaling.
+ resetRecommendedParallelism(evaluatedMetrics.getVertexMetrics());
+ return false;
+ }
+ return true;
+ }
+
+ /** Removes any retained evaluation state for the given job. */
+ public void cleanup(KEY jobKey) {
+ lastEvaluatedMetrics.remove(jobKey);
+ }
+
+ /**
+ * Computes the evaluated scaling metrics for all vertices. The configured custom evaluator
+ * plugin is run only when a non-null context is supplied (the plugin's {@link
+ * ScalingMetricsEvaluatorPlugin.Context} is a thin extension of that canonical context, so it
+ * needs the context to exist). Pass {@code null} for plugin-free computation, as the tests do.
+ */
+ @VisibleForTesting
+ EvaluatedMetrics computeEvaluatedMetrics(
+ @Nullable Context autoScalerContext,
+ Configuration conf,
+ CollectedMetricHistory collectedMetrics,
+ Duration restartTime) {
LOG.debug("Restart time used in metrics evaluation: {}", restartTime);
var scalingOutput = new HashMap<JobVertexID, Map<ScalingMetric, EvaluatedScalingMetric>>();
var metricsHistory = collectedMetrics.getMetricHistory();
@@ -94,22 +153,9 @@ public class ScalingMetricEvaluator {
boolean processingBacklog = isProcessingBacklog(topology, metricsHistory, conf);
- var customEvaluatorWithConfig = getCustomEvaluatorIfRequired(conf);
var customEvaluationSession =
- Optional.ofNullable(customEvaluatorWithConfig)
- .map(
- info ->
- Tuple2.of(
- info.f0,
- new ScalingMetricsEvaluatorPlugin.Context(
- mergeEvaluatorConfig(conf, info.f1),
- Collections.unmodifiableSortedMap(
- metricsHistory),
- Collections.unmodifiableMap(scalingOutput),
- topology,
- processingBacklog,
- restartTime)))
- .orElse(null);
+ buildCustomEvaluationSession(
+ autoScalerContext, conf, scalingOutput, processingBacklog);
for (var vertex : topology.getVerticesInTopologicalOrder()) {
scalingOutput.put(
@@ -129,6 +175,29 @@ public class ScalingMetricEvaluator {
return new EvaluatedMetrics(scalingOutput, globalMetrics);
}
+ @Nullable
+ private Tuple2<ScalingMetricsEvaluatorPlugin, ScalingMetricsEvaluatorPlugin.Context<KEY>>
+ buildCustomEvaluationSession(
+ @Nullable Context autoScalerContext,
+ Configuration conf,
+ Map<JobVertexID, Map<ScalingMetric, EvaluatedScalingMetric>> scalingOutput,
+ boolean processingBacklog) {
+ if (autoScalerContext == null) {
+ return null;
+ }
+ var customEvaluatorWithConfig = getCustomEvaluatorIfRequired(conf);
+ if (customEvaluatorWithConfig == null) {
+ return null;
+ }
+ var pluginContext =
+ new ScalingMetricsEvaluatorPlugin.Context<KEY>(
+ autoScalerContext,
+ customEvaluatorWithConfig.f1,
+ Collections.unmodifiableMap(scalingOutput),
+ processingBacklog);
+ return Tuple2.of(customEvaluatorWithConfig.f0, pluginContext);
+ }
+
@VisibleForTesting
protected static boolean isProcessingBacklog(
JobTopology topology,
@@ -166,7 +235,9 @@ public class ScalingMetricEvaluator {
boolean processingBacklog,
Duration restartTime,
@Nullable
- Tuple2<ScalingMetricsEvaluatorPlugin, ScalingMetricsEvaluatorPlugin.Context>
+ Tuple2<
+ ScalingMetricsEvaluatorPlugin,
+ ScalingMetricsEvaluatorPlugin.Context<KEY>>
customEvaluationSession) {
var latestVertexMetrics =
@@ -694,10 +765,10 @@ public class ScalingMetricEvaluator {
* @return A map of scaling metrics, with its corresponding evaluated scaling metric.
*/
@VisibleForTesting
- protected static Map<ScalingMetric, EvaluatedScalingMetric> runCustomEvaluator(
+ protected static <KEY> Map<ScalingMetric, EvaluatedScalingMetric> runCustomEvaluator(
JobVertexID vertex,
Map<ScalingMetric, EvaluatedScalingMetric> evaluatedMetrics,
- Tuple2<ScalingMetricsEvaluatorPlugin, ScalingMetricsEvaluatorPlugin.Context>
+ Tuple2<ScalingMetricsEvaluatorPlugin, ScalingMetricsEvaluatorPlugin.Context<KEY>>
customEvaluationSession) {
try {
return customEvaluationSession.f0.evaluateVertexMetrics(
@@ -712,24 +783,6 @@ public class ScalingMetricEvaluator {
return Collections.emptyMap();
}
- /**
- * Builds the configuration handed to a custom evaluator: the full job configuration with the
- * evaluator-specific options (prefix already stripped) merged on top, so an evaluator-specific
- * key takes precedence over the job-level value of the same key. The result is an un-modifiable
- * view.
- *
- * @param jobConf The full job configuration.
- * @param evaluatorConf The evaluator-specific options with their prefix already stripped.
- * @return An un-modifiable merged configuration.
- */
- @VisibleForTesting
- protected static Configuration mergeEvaluatorConfig(
- Configuration jobConf, Configuration evaluatorConf) {
- Configuration merged = new Configuration(jobConf);
- merged.addAll(evaluatorConf);
- return new UnmodifiableConfiguration(merged);
- }
-
/**
* Resolves the configured custom metric evaluator (if any) from the registered evaluators.
*
diff --git a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/alignment/BuiltInAlignmentMode.java b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/alignment/BuiltInAlignmentMode.java
index 968bfc1f7..37dcc7310 100644
--- a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/alignment/BuiltInAlignmentMode.java
+++ b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/alignment/BuiltInAlignmentMode.java
@@ -42,7 +42,7 @@ public enum BuiltInAlignmentMode implements ParallelismAlignmentMode, DescribedE
+ "when one is within reach. Balances even data distribution against resource "
+ "usage, tolerating mild skew to avoid over-provisioning.") {
@Override
- public int alignParallelism(Context ctx) {
+ public int alignParallelism(Context<?> ctx) {
return alignOrKeepTarget(ctx, true);
}
},
@@ -51,14 +51,14 @@ public enum BuiltInAlignmentMode implements ParallelismAlignmentMode, DescribedE
"Aligns to a parallelism that evenly divides the number of key groups or source "
+ "partitions, spreading data evenly across subtasks to reduce skew.") {
@Override
- public int alignParallelism(Context ctx) {
+ public int alignParallelism(Context<?> ctx) {
return alignOrKeepTarget(ctx, false);
}
},
OFF("Disables alignment. The autoscaler's computed target parallelism is used as-is.") {
@Override
- public int alignParallelism(Context ctx) {
+ public int alignParallelism(Context<?> ctx) {
return ctx.getNewParallelism();
}
};
@@ -67,7 +67,7 @@ public enum BuiltInAlignmentMode implements ParallelismAlignmentMode, DescribedE
* Scans the direction's region for an aligned value, and keeps the computed target when none is
* found (never blocking the scale).
*/
- private static int alignOrKeepTarget(Context ctx, boolean acceptLoadReducing) {
+ private static int alignOrKeepTarget(Context<?> ctx, boolean acceptLoadReducing) {
int aligned = ParallelismAligner.firstAlignedInRegion(ctx, acceptLoadReducing);
return aligned > 0 ? aligned : ctx.getNewParallelism();
}
diff --git a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/alignment/KeyGroupOrPartitionsAdjustMode.java b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/alignment/KeyGroupOrPartitionsAdjustMode.java
index 8f1fbd4ae..85ef8dabd 100644
--- a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/alignment/KeyGroupOrPartitionsAdjustMode.java
+++ b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/alignment/KeyGroupOrPartitionsAdjustMode.java
@@ -59,13 +59,13 @@ public enum KeyGroupOrPartitionsAdjustMode implements ParallelismAlignmentMode,
}
@Override
- public int alignParallelism(Context ctx) {
+ public int alignParallelism(Context<?> ctx) {
// Called directly (without an emitter) the SCALING_LIMITED event is suppressed; the
// autoscaler path goes through the emitter overload below via ParallelismAligner.
return alignParallelism(ctx, (expected, actual) -> {});
}
- int alignParallelism(Context ctx, ParallelismAligner.ScalingLimitedEmitter emitter) {
+ int alignParallelism(Context<?> ctx, ParallelismAligner.ScalingLimitedEmitter emitter) {
int aligned = ParallelismAligner.firstAlignedInRegion(ctx, maximize);
if (aligned > 0) {
return aligned;
diff --git a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/alignment/ParallelismAligner.java b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/alignment/ParallelismAligner.java
index 4ca14a9cd..b58da779b 100644
--- a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/alignment/ParallelismAligner.java
+++ b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/alignment/ParallelismAligner.java
@@ -23,7 +23,6 @@ import org.apache.flink.autoscaler.config.AutoScalerOptions;
import org.apache.flink.autoscaler.event.AutoScalerEventHandler;
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.ShipStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.jobgraph.JobVertexID;
@@ -90,14 +89,15 @@ public final class ParallelismAligner {
int parallelismUpperLimit,
Collection<ShipStrategy> inputShipStrategies,
Map<ScalingMetric, EvaluatedScalingMetric> evaluatedMetrics,
- JobTopology jobTopology,
AutoScalerEventHandler<KEY, Context> eventHandler,
Context context) {
Configuration conf = context.getConfiguration();
String modeName = conf.get(ALIGNMENT_MODE);
- ParallelismAlignmentMode.Context alignmentContext =
- new ParallelismAlignmentMode.Context(
+ ParallelismAlignmentMode.Context<KEY> alignmentContext =
+ new ParallelismAlignmentMode.Context<>(
+ context,
+ AutoScalerOptions.customAlignmentModeConfiguration(conf, modeName),
vertex,
currentParallelism,
newParallelism,
@@ -106,10 +106,7 @@ public final class ParallelismAligner {
parallelismLowerLimit,
parallelismUpperLimit,
inputShipStrategies,
- context,
- evaluatedMetrics,
- jobTopology,
- AutoScalerOptions.customAlignmentModeConfiguration(conf, modeName));
+ evaluatedMetrics);
ParallelismAlignmentMode mode = resolve(conf);
if (!mode.isApplicable(alignmentContext)) {
@@ -193,19 +190,19 @@ public final class ParallelismAligner {
}
/** Whether the computed target is above the current parallelism. */
- public static boolean isScaleUp(ParallelismAlignmentMode.Context ctx) {
+ public static boolean isScaleUp(ParallelismAlignmentMode.Context<?> ctx) {
return ctx.getNewParallelism() > ctx.getCurrentParallelism();
}
/** {@code N}: the number of key groups or source partitions to align to. */
- public static int numKeyGroupsOrPartitions(ParallelismAlignmentMode.Context ctx) {
+ public static int numKeyGroupsOrPartitions(ParallelismAlignmentMode.Context<?> ctx) {
return ctx.getNumSourcePartitions() <= 0
? ctx.getMaxParallelism()
: ctx.getNumSourcePartitions();
}
/** The alignment cap: {@code min(N, min(maxParallelism, parallelismUpperLimit))}. */
- public static int upperBoundForAlignment(ParallelismAlignmentMode.Context ctx) {
+ public static int upperBoundForAlignment(ParallelismAlignmentMode.Context<?> ctx) {
return Math.min(
numKeyGroupsOrPartitions(ctx),
Math.min(ctx.getMaxParallelism(), ctx.getParallelismUpperLimit()));
@@ -225,7 +222,7 @@ public final class ParallelismAligner {
* @return the first accepted parallelism, or {@code 0} when none is found in the region
*/
public static int firstAlignedInRegion(
- ParallelismAlignmentMode.Context ctx, boolean acceptLoadReducing) {
+ ParallelismAlignmentMode.Context<?> ctx, boolean acceptLoadReducing) {
int n = numKeyGroupsOrPartitions(ctx);
int target = ctx.getNewParallelism();
int regionEnd =
@@ -247,7 +244,7 @@ public final class ParallelismAligner {
* where per-subtask load increases, snapping up to the nearest divisor, then clamps to the
* parallelism lower limit.
*/
- public static int relaxedDownwardFallback(ParallelismAlignmentMode.Context ctx) {
+ public static int relaxedDownwardFallback(ParallelismAlignmentMode.Context<?> ctx) {
int n = numKeyGroupsOrPartitions(ctx);
int target = ctx.getNewParallelism();
@@ -270,7 +267,7 @@ public final class ParallelismAligner {
* the aligned value deviates from the computed target. Matches the historical behavior.
*/
public static int applyBlockingFallback(
- ParallelismAlignmentMode.Context ctx, int candidate, ScalingLimitedEmitter emitter) {
+ ParallelismAlignmentMode.Context<?> ctx, int candidate, ScalingLimitedEmitter emitter) {
if (invertsDirection(ctx, candidate)) {
emitter.emit(ctx.getNewParallelism(), ctx.getCurrentParallelism());
return ctx.getCurrentParallelism();
@@ -281,7 +278,8 @@ public final class ParallelismAligner {
return candidate;
}
- private static boolean invertsDirection(ParallelismAlignmentMode.Context ctx, int candidate) {
+ private static boolean invertsDirection(
+ ParallelismAlignmentMode.Context<?> ctx, int candidate) {
return (isScaleUp(ctx) && candidate <= ctx.getCurrentParallelism())
|| (!isScaleUp(ctx) && candidate >= ctx.getCurrentParallelism());
}
diff --git a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/alignment/ParallelismAlignmentMode.java b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/alignment/ParallelismAlignmentMode.java
index 4d4241f4c..bbda7a95a 100644
--- a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/alignment/ParallelismAlignmentMode.java
+++ b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/alignment/ParallelismAlignmentMode.java
@@ -21,12 +21,11 @@ import org.apache.flink.annotation.Experimental;
import org.apache.flink.autoscaler.JobAutoScalerContext;
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.ShipStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.jobgraph.JobVertexID;
-import lombok.Value;
+import lombok.Getter;
import java.util.Collection;
import java.util.Map;
@@ -64,7 +63,7 @@ public interface ParallelismAlignmentMode {
* partitioned sources that report a partition count (Kafka and Pulsar do by default). A custom
* mode can widen this, for example to align custom partitioned vertices.
*/
- default boolean isApplicable(Context ctx) {
+ default boolean isApplicable(Context<?> ctx) {
return ctx.getNumSourcePartitions() > 0
|| ctx.getInputShipStrategies().contains(ShipStrategy.HASH);
}
@@ -73,50 +72,71 @@ public interface ParallelismAlignmentMode {
* Returns the parallelism to apply for the vertex described by {@code ctx}. Called only when
* {@link #isApplicable(Context)} returned {@code true}.
*/
- int alignParallelism(Context ctx);
+ int alignParallelism(Context<?> ctx);
/**
- * Immutable inputs to a single per-vertex parallelism alignment, handed to {@link
- * ParallelismAlignmentMode#alignParallelism(Context)}. Besides the parallelism inputs it
- * exposes the full autoscaler context, the per-vertex evaluated metrics, the job topology, and
- * a prefix-stripped configuration for the selected mode, so custom modes have what they need.
+ * The parallelism alignment context. It {@code extends} {@link JobAutoScalerContext}, sharing
+ * its {@link org.apache.flink.autoscaler.JobAutoScalerContext.ScalingCycleState} and inherited
+ * cycle accessors (such as {@link #getJobTopology()} and {@link #getEvaluatedMetrics()}). Its
+ * {@code getConfiguration()} returns the effective per-mode configuration: the job
+ * configuration with this mode's prefix-stripped {@code
+ * scaling.parallelism-alignment.mode.<name>.} overrides merged on top. It adds only the
+ * per-vertex alignment inputs that are not already available on the canonical context.
+ *
+ * @param <KEY> The job key.
*/
- @Value
- class Context {
+ @Getter
+ class Context<KEY> extends JobAutoScalerContext<KEY> {
+
/** The vertex being aligned. */
- JobVertexID vertex;
+ private final JobVertexID vertex;
/** The current parallelism of the vertex. */
- int currentParallelism;
+ private final int currentParallelism;
/** The clamped computed target parallelism, before alignment. */
- int newParallelism;
+ private final int newParallelism;
/** The number of source partitions, or a non-positive value when not a source. */
- int numSourcePartitions;
+ private final int numSourcePartitions;
/** The vertex max parallelism (number of key groups). */
- int maxParallelism;
+ private final int maxParallelism;
/** The configured parallelism lower limit. */
- int parallelismLowerLimit;
+ private final int parallelismLowerLimit;
/** The configured parallelism upper limit. */
- int parallelismUpperLimit;
+ private final int parallelismUpperLimit;
/** The ship strategies of the vertex inputs. */
- Collection<ShipStrategy> inputShipStrategies;
-
- /** The full autoscaler context (cluster client, job configuration, etc.). */
- JobAutoScalerContext<?> jobContext;
-
- /** The per-vertex evaluated metrics. */
- Map<ScalingMetric, EvaluatedScalingMetric> evaluatedMetrics;
-
- /** The job topology. */
- JobTopology jobTopology;
-
- /** The prefix-stripped configuration for the selected (custom) mode. */
- Configuration modeConfiguration;
+ private final Collection<ShipStrategy> inputShipStrategies;
+
+ /** The evaluated metrics of the vertex being aligned. */
+ private final Map<ScalingMetric, EvaluatedScalingMetric> evaluatedVertexMetrics;
+
+ public Context(
+ JobAutoScalerContext<KEY> autoScalerContext,
+ Configuration overrides,
+ JobVertexID vertex,
+ int currentParallelism,
+ int newParallelism,
+ int numSourcePartitions,
+ int maxParallelism,
+ int parallelismLowerLimit,
+ int parallelismUpperLimit,
+ Collection<ShipStrategy> inputShipStrategies,
+ Map<ScalingMetric, EvaluatedScalingMetric> evaluatedVertexMetrics) {
+ super(autoScalerContext, overrides);
+ this.vertex = vertex;
+ this.currentParallelism = currentParallelism;
+ this.newParallelism = newParallelism;
+ this.numSourcePartitions = numSourcePartitions;
+ this.maxParallelism = maxParallelism;
+ this.parallelismLowerLimit = parallelismLowerLimit;
+ this.parallelismUpperLimit = parallelismUpperLimit;
+ this.inputShipStrategies = inputShipStrategies;
+ this.evaluatedVertexMetrics = evaluatedVertexMetrics;
+ }
}
}
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 291e9fabe..8dc99c503 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
@@ -24,6 +24,7 @@ import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.MemorySize;
+import org.apache.flink.configuration.UnmodifiableConfiguration;
import java.time.Duration;
import java.util.List;
@@ -435,7 +436,7 @@ public class AutoScalerOptions {
"An arbitrary parameter passed to the custom alignment mode named "
+ "<name>. All such parameters are made available to the mode "
+ "(with the prefix stripped) through "
- + "Context.getModeConfiguration().");
+ + "Context.getConfiguration().");
public static final ConfigOption<Boolean> OBSERVED_SCALABILITY_ENABLED =
autoScalerConfig("observed-scalability.enabled")
@@ -672,6 +673,19 @@ public class AutoScalerOptions {
return prefixStrippedConfiguration(conf, SCALING_ALIGNMENT_MODE_CONF_PREFIX, name);
}
+ /**
+ * Overlays the prefix-stripped per-instance plugin {@code overrides} onto {@code base} and
+ * returns an unmodifiable view. Values from {@code overrides} take precedence on overlap.
+ * Shared by the custom evaluator, scaling executor and parallelism alignment plugins so the
+ * merge semantics stay consistent across all three.
+ */
+ public static UnmodifiableConfiguration overlayConfiguration(
+ Configuration base, Configuration overrides) {
+ Configuration merged = new Configuration(base);
+ merged.addAll(overrides);
+ return new UnmodifiableConfiguration(merged);
+ }
+
/**
* Builds a fresh {@link Configuration} exposing the per-instance options configured under
* {@code job.autoscaler.<confPrefix><name>.} (canonical) and {@code
diff --git a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingMetricsEvaluatorPlugin.java b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingMetricsEvaluatorPlugin.java
index 59d9978de..c07450399 100644
--- a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingMetricsEvaluatorPlugin.java
+++ b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingMetricsEvaluatorPlugin.java
@@ -19,16 +19,13 @@
package org.apache.flink.autoscaler.metrics;
import org.apache.flink.annotation.Experimental;
-import org.apache.flink.autoscaler.topology.JobTopology;
+import org.apache.flink.autoscaler.JobAutoScalerContext;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.jobgraph.JobVertexID;
-import lombok.Value;
+import lombok.Getter;
-import java.time.Duration;
-import java.time.Instant;
import java.util.Map;
-import java.util.SortedMap;
/**
* A pluggable plugin that allows users to provide custom scaling-metric evaluation logic on top of
@@ -69,46 +66,44 @@ public interface ScalingMetricsEvaluatorPlugin {
* @return A map of evaluated scaling metrics for the vertex which would get merged with
* internally evaluated metrics.
* @throws UnsupportedOperationException if an attempt is made to modify the {@code
- * evaluatedMetrics}, {@code Context.config}, {@code Context.metricsHistory}, {@code
- * Context.evaluatedVertexMetrics}.
+ * evaluatedMetrics}, the context's configuration, its metric history or its evaluated
+ * vertex metrics.
*/
Map<ScalingMetric, EvaluatedScalingMetric> evaluateVertexMetrics(
JobVertexID vertex,
Map<ScalingMetric, EvaluatedScalingMetric> evaluatedMetrics,
- Context evaluationContext);
+ Context<?> evaluationContext);
/**
- * Context providing relevant job and metric information to assist in custom metric evaluation.
+ * The custom metric evaluator context. It {@code extends} {@link JobAutoScalerContext}, sharing
+ * its {@link org.apache.flink.autoscaler.JobAutoScalerContext.ScalingCycleState} and inherited
+ * cycle accessors (topology, metric history, restart time). Its {@code getConfiguration()}
+ * returns the effective evaluator configuration (the job configuration with this evaluator's
+ * {@code job.autoscaler.metrics.custom-evaluator.<name>.} overrides merged on top), enriched
+ * with the in-progress evaluated vertex metrics ({@code getEvaluatedVertexMetrics()}) and the
+ * backlog flag ({@code isProcessingBacklog()}).
*/
- @Value
- class Context {
- /**
- * An un-modifiable view of the job configuration, with the evaluator-specific options
- * (configured under {@code job.autoscaler.metrics.custom-evaluator.<name>.}) merged on top
- * with their prefix stripped. Evaluator-specific keys take precedence over the job-level
- * value of the same key, so a CR can override behaviour for this evaluator alone.
- */
- Configuration config;
-
- /**
- * An un-modifiable view of the historical record of collected metrics, ordered by
- * timestamp.
- */
- SortedMap<Instant, CollectedMetrics> metricsHistory;
+ @Getter
+ class Context<KEY> extends JobAutoScalerContext<KEY> {
/**
* An un-modifiable view of evaluated metrics for previously evaluated vertices. Note:
* evaluation of a vertex for scaling metrics happens topologically.
*/
- Map<JobVertexID, Map<ScalingMetric, EvaluatedScalingMetric>> evaluatedVertexMetrics;
-
- /** The job topology representing the structure of the Flink job. */
- JobTopology topology;
+ private final Map<JobVertexID, Map<ScalingMetric, EvaluatedScalingMetric>>
+ evaluatedVertexMetrics;
/** Indicates whether the job is processing backlog. */
- boolean processingBacklog;
+ private final boolean processingBacklog;
- /** Maximum restart time based on scaling records. */
- Duration restartTime;
+ public Context(
+ JobAutoScalerContext<KEY> autoScalerContext,
+ Configuration overrides,
+ Map<JobVertexID, Map<ScalingMetric, EvaluatedScalingMetric>> evaluatedVertexMetrics,
+ boolean processingBacklog) {
+ super(autoScalerContext, overrides);
+ this.evaluatedVertexMetrics = evaluatedVertexMetrics;
+ this.processingBacklog = processingBacklog;
+ }
}
}
diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/AutoScalerCustomEvaluatorTest.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/AutoScalerCustomEvaluatorTest.java
index a7b5418cf..2137b54e0 100644
--- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/AutoScalerCustomEvaluatorTest.java
+++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/AutoScalerCustomEvaluatorTest.java
@@ -90,7 +90,8 @@ public class AutoScalerCustomEvaluatorTest {
Map.of(source1, REBALANCE),
1,
720,
- new IOMetrics(0, 0, 0))));
+ new IOMetrics(0, 0, 0))),
+ stateStore);
var defaultConf = context.getConfiguration();
defaultConf.set(AutoScalerOptions.AUTOSCALER_ENABLED, true);
@@ -117,7 +118,7 @@ public class AutoScalerCustomEvaluatorTest {
autoscaler =
new JobAutoScalerImpl<>(
metricsCollector,
- new ScalingMetricEvaluator(customEvaluators),
+ new ScalingMetricEvaluator<>(customEvaluators),
scalingExecutor,
eventCollector,
new TestingScalingRealizer<>(),
diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/BacklogBasedScalingTest.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/BacklogBasedScalingTest.java
index 9ff58ab21..9355806e6 100644
--- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/BacklogBasedScalingTest.java
+++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/BacklogBasedScalingTest.java
@@ -86,7 +86,8 @@ public class BacklogBasedScalingTest {
Map.of(source1, REBALANCE),
1,
720,
- new IOMetrics(0, 0, 0))));
+ new IOMetrics(0, 0, 0))),
+ stateStore);
var defaultConf = context.getConfiguration();
defaultConf.set(AutoScalerOptions.AUTOSCALER_ENABLED, true);
@@ -105,7 +106,7 @@ public class BacklogBasedScalingTest {
autoscaler =
new JobAutoScalerImpl<>(
metricsCollector,
- new ScalingMetricEvaluator(List.of()),
+ new ScalingMetricEvaluator<>(List.of()),
scalingExecutor,
eventCollector,
new TestingScalingRealizer<>(),
@@ -406,7 +407,7 @@ public class BacklogBasedScalingTest {
setClocksTo(now);
metricsCollector.setJobUpdateTs(now);
autoscaler.scale(context);
- assertTrue(autoscaler.lastEvaluatedMetrics.isEmpty());
+ assertTrue(autoscaler.metricsEvaluator.lastEvaluatedMetrics.isEmpty());
assertTrue(eventCollector.events.isEmpty());
}
diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/DelayedScaleDownEndToEndTest.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/DelayedScaleDownEndToEndTest.java
index 43c9f8a1b..d35fcf29e 100644
--- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/DelayedScaleDownEndToEndTest.java
+++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/DelayedScaleDownEndToEndTest.java
@@ -92,7 +92,8 @@ public class DelayedScaleDownEndToEndTest {
sink,
Map.of(source, REBALANCE),
INITIAL_SINK_PARALLELISM,
- 4000)));
+ 4000)),
+ stateStore);
var scaleDownInterval = Duration.ofMinutes(60).minus(Duration.ofSeconds(1));
// The metric window size is 9:59 to avoid other metrics are mixed.
@@ -118,14 +119,15 @@ public class DelayedScaleDownEndToEndTest {
autoscaler =
new JobAutoScalerImpl<>(
metricsCollector,
- new ScalingMetricEvaluator(List.of()),
+ new ScalingMetricEvaluator<>(List.of()),
new ScalingExecutor<>(eventCollector, stateStore),
eventCollector,
scalingRealizer,
stateStore);
// initially the last evaluated metrics are empty
- assertThat(autoscaler.lastEvaluatedMetrics.get(context.getJobKey())).isNull();
+ assertThat(autoscaler.metricsEvaluator.lastEvaluatedMetrics.get(context.getJobKey()))
+ .isNull();
now = Instant.ofEpochMilli(0);
setClocksTo(now);
@@ -688,6 +690,7 @@ public class DelayedScaleDownEndToEndTest {
private Double getCurrentMetricValue(JobVertexID jobVertexID, ScalingMetric scalingMetric) {
var metric =
autoscaler
+ .metricsEvaluator
.lastEvaluatedMetrics
.get(context.getJobKey())
.getVertexMetrics()
diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/JobAutoScalerImplTest.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/JobAutoScalerImplTest.java
index 384d97631..7eac783aa 100644
--- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/JobAutoScalerImplTest.java
+++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/JobAutoScalerImplTest.java
@@ -92,7 +92,8 @@ public class JobAutoScalerImplTest {
JobTopology jobTopology = new JobTopology(new VertexInfo(jobVertexID, Map.of(), 1, 10));
var metricsCollector =
- new TestingMetricsCollector<JobID, JobAutoScalerContext<JobID>>(jobTopology);
+ new TestingMetricsCollector<JobID, JobAutoScalerContext<JobID>>(
+ jobTopology, stateStore);
metricsCollector.updateMetrics(
jobVertexID,
TestMetrics.builder()
@@ -103,7 +104,8 @@ public class JobAutoScalerImplTest {
.pendingRecords(0L)
.build());
- ScalingMetricEvaluator evaluator = new ScalingMetricEvaluator(Collections.emptyList());
+ ScalingMetricEvaluator<JobID, JobAutoScalerContext<JobID>> evaluator =
+ new ScalingMetricEvaluator<>(Collections.emptyList());
ScalingExecutor<JobID, JobAutoScalerContext<JobID>> scalingExecutor =
new ScalingExecutor<>(eventCollector, stateStore);
@@ -170,7 +172,8 @@ public class JobAutoScalerImplTest {
public void testTolerateRecoverableExceptions() throws Exception {
TestingMetricsCollector<JobID, JobAutoScalerContext<JobID>>
collectorWhichThrowsRecoverableException =
- new TestingMetricsCollector<>(new JobTopology(Collections.emptySet())) {
+ new TestingMetricsCollector<>(
+ new JobTopology(Collections.emptySet()), stateStore) {
@Override
protected Collection<String> queryAggregatedMetricNames(
RestClusterClient<?> restClient,
@@ -201,7 +204,8 @@ public class JobAutoScalerImplTest {
JobVertexID jobVertexID = new JobVertexID();
JobTopology jobTopology = new JobTopology(new VertexInfo(jobVertexID, Map.of(), 1, 20));
var metricsCollector =
- new TestingMetricsCollector<JobID, JobAutoScalerContext<JobID>>(jobTopology);
+ new TestingMetricsCollector<JobID, JobAutoScalerContext<JobID>>(
+ jobTopology, stateStore);
ScalingRealizer<JobID, JobAutoScalerContext<JobID>>
realizeParallelismOverridesWithExceptionsScalingRealizer =
new ScalingRealizer<>() {
@@ -239,8 +243,8 @@ public class JobAutoScalerImplTest {
void testParallelismOverrides() throws Exception {
var autoscaler =
new JobAutoScalerImpl<>(
- new TestingMetricsCollector<>(new JobTopology()),
- null,
+ new TestingMetricsCollector<>(new JobTopology(), stateStore),
+ new ScalingMetricEvaluator<>(Collections.emptyList()),
null,
eventCollector,
scalingRealizer,
@@ -383,8 +387,8 @@ public class JobAutoScalerImplTest {
var notRunningContext = context.toBuilder().jobStatus(JobStatus.INITIALIZING).build();
var autoscaler =
new JobAutoScalerImpl<>(
- new TestingMetricsCollector<>(new JobTopology()),
- null,
+ new TestingMetricsCollector<>(new JobTopology(), stateStore),
+ new ScalingMetricEvaluator<>(Collections.emptyList()),
null,
eventCollector,
scalingRealizer,
diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/MetricsCollectionAndEvaluationTest.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/MetricsCollectionAndEvaluationTest.java
index c347cff8a..64e8f1772 100644
--- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/MetricsCollectionAndEvaluationTest.java
+++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/MetricsCollectionAndEvaluationTest.java
@@ -63,7 +63,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
public class MetricsCollectionAndEvaluationTest {
private JobAutoScalerContext<JobID> context;
- private ScalingMetricEvaluator evaluator;
+ private ScalingMetricEvaluator<JobID, JobAutoScalerContext<JobID>> evaluator;
private TestingMetricsCollector<JobID, JobAutoScalerContext<JobID>> metricsCollector;
private ScalingExecutor<JobID, JobAutoScalerContext<JobID>> scalingExecutor;
private InMemoryAutoScalerStateStore<JobID, JobAutoScalerContext<JobID>> stateStore;
@@ -80,7 +80,7 @@ public class MetricsCollectionAndEvaluationTest {
@BeforeEach
public void setup() {
context = createDefaultJobAutoScalerContext();
- evaluator = new ScalingMetricEvaluator(List.of());
+ evaluator = new ScalingMetricEvaluator<>(List.of());
stateStore = new InMemoryAutoScalerStateStore<>();
scalingExecutor = new ScalingExecutor<>(new TestingEventCollector<>(), stateStore);
@@ -103,7 +103,7 @@ public class MetricsCollectionAndEvaluationTest {
new VertexInfo(
sink, Map.of(map, REBALANCE), 8, 24, new IOMetrics(0, 0, 0)));
- metricsCollector = new TestingMetricsCollector<>(topology);
+ metricsCollector = new TestingMetricsCollector<>(topology, stateStore);
var conf = context.getConfiguration();
conf.set(AutoScalerOptions.STABILIZATION_INTERVAL, Duration.ofSeconds(10));
@@ -131,14 +131,14 @@ public class MetricsCollectionAndEvaluationTest {
// We haven't left the stabilization period, but we're collecting and returning the metrics
// for reporting
- var collectedMetrics = metricsCollector.updateMetrics(context, stateStore);
+ var collectedMetrics = collect();
assertEquals(1, collectedMetrics.getMetricHistory().size());
// We haven't collected a full window yet, but we're collecting and returning the metrics
// for reporting
clock = Clock.offset(clock, conf.get(AutoScalerOptions.STABILIZATION_INTERVAL));
metricsCollector.setClock(clock);
- collectedMetrics = metricsCollector.updateMetrics(context, stateStore);
+ collectedMetrics = collect();
assertEquals(2, collectedMetrics.getMetricHistory().size());
assertFalse(collectedMetrics.isFullyCollected());
@@ -147,7 +147,7 @@ public class MetricsCollectionAndEvaluationTest {
clock = Clock.offset(clock, Duration.ofSeconds(1));
metricsCollector.setClock(clock);
- collectedMetrics = metricsCollector.updateMetrics(context, stateStore);
+ collectedMetrics = collect();
assertEquals(3, collectedMetrics.getMetricHistory().size());
assertFalse(collectedMetrics.isFullyCollected());
@@ -159,13 +159,13 @@ public class MetricsCollectionAndEvaluationTest {
.plus(conf.get(AutoScalerOptions.METRICS_WINDOW)),
ZoneId.systemDefault());
metricsCollector.setClock(clock);
- collectedMetrics = metricsCollector.updateMetrics(context, stateStore);
+ collectedMetrics = collect();
assertEquals(3, collectedMetrics.getMetricHistory().size());
assertTrue(collectedMetrics.isFullyCollected());
// Test resetting the collector and make sure we can deserialize the scalingInfo correctly
- metricsCollector = new TestingMetricsCollector<>(topology);
+ metricsCollector = new TestingMetricsCollector<>(topology, stateStore);
metricsCollector.setJobUpdateTs(startTime);
metricsCollector.setClock(clock);
setDefaultMetrics(metricsCollector);
@@ -177,11 +177,12 @@ public class MetricsCollectionAndEvaluationTest {
metricsCollector.updateMetrics(map, tm -> tm.setNumRecordsIn(2000));
metricsCollector.updateMetrics(sink, tm -> tm.setNumRecordsIn(4000));
- collectedMetrics = metricsCollector.updateMetrics(context, stateStore);
+ collectedMetrics = collect();
assertEquals(3, collectedMetrics.getMetricHistory().size());
assertTrue(collectedMetrics.isFullyCollected());
- var evaluation = evaluator.evaluate(conf, collectedMetrics, restartTime);
+ var evaluation =
+ evaluator.computeEvaluatedMetrics(null, conf, collectedMetrics, restartTime);
scalingExecutor.scaleResource(
context,
evaluation,
@@ -236,12 +237,12 @@ public class MetricsCollectionAndEvaluationTest {
@Test
public void testKafkaPulsarNumPartitions() throws Exception {
setDefaultMetrics(metricsCollector);
- metricsCollector.updateMetrics(context, stateStore);
+ collect();
var clock = Clock.fixed(Instant.now().plus(Duration.ofSeconds(3)), ZoneId.systemDefault());
metricsCollector.setClock(clock);
- var collectedMetrics = metricsCollector.updateMetrics(context, stateStore);
+ var collectedMetrics = collect();
clock = Clock.fixed(Instant.now().plus(Duration.ofSeconds(3)), ZoneId.systemDefault());
metricsCollector.setClock(clock);
metricsCollector.setMetricNames(
@@ -255,7 +256,7 @@ public class MetricsCollectionAndEvaluationTest {
"1.Source__Kafka_Source_(testTopic).KafkaSourceReader.topic.testTopic.partition.2.currentOffset",
"1.Source__Kafka_Source_(testTopic).KafkaSourceReader.topic.testTopic.partition.3.currentOffset")));
- collectedMetrics = metricsCollector.updateMetrics(context, stateStore);
+ collectedMetrics = collect();
assertEquals(5, collectedMetrics.getJobTopology().get(source1).getNumSourcePartitions());
metricsCollector.setMetricNames(
@@ -269,7 +270,7 @@ public class MetricsCollectionAndEvaluationTest {
"1.Source__Kafka_Source_(testTopic).kafkaCluster.my-cluster-2.KafkaSourceReader.topic.testTopic.partition.2.currentOffset",
"1.Source__Kafka_Source_(testTopic).kafkaCluster.my-cluster-2.KafkaSourceReader.topic.testTopic.partition.3.currentOffset")));
- collectedMetrics = metricsCollector.updateMetrics(context, stateStore);
+ collectedMetrics = collect();
assertEquals(6, collectedMetrics.getJobTopology().get(source1).getNumSourcePartitions());
metricsCollector.setMetricNames(
@@ -288,7 +289,7 @@ public class MetricsCollectionAndEvaluationTest {
+ ".persistent_//public/default/testTopic-partition-3.e427h.numMsgsReceived",
"0.Source__pulsar_source[1].PulsarConsumer"
+ ".persistent_//public/default/testTopic-partition-4.m962n.numMsgsReceived")));
- collectedMetrics = metricsCollector.updateMetrics(context, stateStore);
+ collectedMetrics = collect();
assertEquals(5, collectedMetrics.getJobTopology().get(source2).getNumSourcePartitions());
}
@@ -307,12 +308,12 @@ public class MetricsCollectionAndEvaluationTest {
@Test
public void testMetricCollectorWindow() throws Exception {
setDefaultMetrics(metricsCollector);
- var metricsHistory = metricsCollector.updateMetrics(context, stateStore);
+ var metricsHistory = collect();
assertEquals(1, metricsHistory.getMetricHistory().size());
// Not stable, metrics collected and reported
metricsCollector.setClock(Clock.offset(clock, Duration.ofSeconds(1)));
- metricsHistory = metricsCollector.updateMetrics(context, stateStore);
+ metricsHistory = collect();
assertEquals(2, metricsHistory.getMetricHistory().size());
// Update clock to stable time
@@ -322,37 +323,37 @@ public class MetricsCollectionAndEvaluationTest {
// This call will lead to metric collection but we haven't reached the window size yet
// which will hold back metrics
- metricsHistory = metricsCollector.updateMetrics(context, stateStore);
+ metricsHistory = collect();
assertEquals(3, metricsHistory.getMetricHistory().size());
assertFalse(metricsHistory.isFullyCollected());
// Collect more values in window
metricsCollector.setClock(Clock.offset(clock, Duration.ofSeconds(1)));
- metricsHistory = metricsCollector.updateMetrics(context, stateStore);
+ metricsHistory = collect();
assertEquals(4, metricsHistory.getMetricHistory().size());
assertFalse(metricsHistory.isFullyCollected());
// Window size reached
metricsCollector.setClock(Clock.offset(clock, conf.get(AutoScalerOptions.METRICS_WINDOW)));
- metricsHistory = metricsCollector.updateMetrics(context, stateStore);
+ metricsHistory = collect();
assertEquals(3, metricsHistory.getMetricHistory().size());
assertTrue(metricsHistory.isFullyCollected());
// Window size + 1 will invalidate the first metric
metricsCollector.setClock(
Clock.offset(clock, conf.get(AutoScalerOptions.METRICS_WINDOW).plusSeconds(1)));
- metricsHistory = metricsCollector.updateMetrics(context, stateStore);
+ metricsHistory = collect();
assertEquals(3, metricsHistory.getMetricHistory().size());
// Complete new metric window with just the currently collected metric
metricsCollector.setClock(
Clock.offset(clock, conf.get(AutoScalerOptions.METRICS_WINDOW).plusDays(1)));
- metricsHistory = metricsCollector.updateMetrics(context, stateStore);
+ metricsHistory = collect();
assertEquals(1, metricsHistory.getMetricHistory().size());
// Existing metrics should be cleared on job updates, and start collecting from fresh
metricsCollector.setJobUpdateTs(clock.instant().plus(Duration.ofDays(10)));
- metricsHistory = metricsCollector.updateMetrics(context, stateStore);
+ metricsHistory = collect();
assertEquals(1, metricsHistory.getMetricHistory().size());
}
@@ -367,7 +368,7 @@ public class MetricsCollectionAndEvaluationTest {
// We haven't left the stabilization period, we're returning the metrics
// but don't act on it since the metric window is not full yet
- var collectedMetrics = metricsCollector.updateMetrics(context, stateStore);
+ var collectedMetrics = collect();
assertFalse(collectedMetrics.getMetricHistory().isEmpty());
assertFalse(collectedMetrics.isFullyCollected());
}
@@ -376,7 +377,7 @@ public class MetricsCollectionAndEvaluationTest {
public void testTolerateAbsenceOfPendingRecordsMetric() throws Exception {
var topology = new JobTopology(new VertexInfo(source1, Map.of(), 5, 720));
- metricsCollector = new TestingMetricsCollector<>(topology);
+ metricsCollector = new TestingMetricsCollector<>(topology, stateStore);
metricsCollector.setJobUpdateTs(startTime);
metricsCollector.updateMetrics(
@@ -394,14 +395,15 @@ public class MetricsCollectionAndEvaluationTest {
metricsCollector.setClock(clock);
- metricsCollector.updateMetrics(context, stateStore);
+ collect();
metricsCollector.setClock(Clock.offset(clock, Duration.ofSeconds(2)));
metricsCollector.updateMetrics(source1, tm -> tm.setNumRecordsIn(1000));
- var collectedMetrics = metricsCollector.updateMetrics(context, stateStore);
+ var collectedMetrics = collect();
var evaluation =
- evaluator.evaluate(context.getConfiguration(), collectedMetrics, restartTime);
+ evaluator.computeEvaluatedMetrics(
+ null, context.getConfiguration(), collectedMetrics, restartTime);
assertEquals(
500.,
evaluation
@@ -459,7 +461,7 @@ public class MetricsCollectionAndEvaluationTest {
true,
IOMetrics.FINISHED_METRICS));
- metricsCollector = new TestingMetricsCollector(topology);
+ metricsCollector = new TestingMetricsCollector(topology, stateStore);
metricsCollector.setJobUpdateTs(startTime);
metricsCollector.updateMetrics(
s1,
@@ -494,7 +496,7 @@ public class MetricsCollectionAndEvaluationTest {
var source = new JobVertexID();
var topology = new JobTopology(new VertexInfo(source, Map.of(), 10, 720));
- metricsCollector = new TestingMetricsCollector(topology);
+ metricsCollector = new TestingMetricsCollector(topology, stateStore);
metricsCollector.setJobUpdateTs(startTime);
metricsCollector.updateMetrics(
source,
@@ -509,8 +511,7 @@ public class MetricsCollectionAndEvaluationTest {
context.getConfiguration().set(AutoScalerOptions.STABILIZATION_INTERVAL, Duration.ZERO);
metricsCollector.setClock(Clock.fixed(Instant.ofEpochMilli(100), ZoneId.systemDefault()));
var collectedMetrics =
- metricsCollector
- .updateMetrics(context, stateStore)
+ collect()
.getMetricHistory()
.get(Instant.ofEpochMilli(100))
.getVertexMetrics()
@@ -523,8 +524,7 @@ public class MetricsCollectionAndEvaluationTest {
metricsCollector.updateMetrics(source, tm -> tm.setPendingRecords(0L));
metricsCollector.setClock(Clock.fixed(Instant.ofEpochMilli(200), ZoneId.systemDefault()));
collectedMetrics =
- metricsCollector
- .updateMetrics(context, stateStore)
+ collect()
.getMetricHistory()
.get(Instant.ofEpochMilli(200))
.getVertexMetrics()
@@ -540,8 +540,7 @@ public class MetricsCollectionAndEvaluationTest {
tm -> tm.setAvgBackpressureTimePerSec(500));
metricsCollector.setClock(Clock.fixed(Instant.ofEpochMilli(300), ZoneId.systemDefault()));
collectedMetrics =
- metricsCollector
- .updateMetrics(context, stateStore)
+ collect()
.getMetricHistory()
.get(Instant.ofEpochMilli(300))
.getVertexMetrics()
@@ -553,8 +552,7 @@ public class MetricsCollectionAndEvaluationTest {
metricsCollector.setClock(Clock.fixed(Instant.ofEpochMilli(400), ZoneId.systemDefault()));
collectedMetrics =
- metricsCollector
- .updateMetrics(context, stateStore)
+ collect()
.getMetricHistory()
.get(Instant.ofEpochMilli(400))
.getVertexMetrics()
@@ -570,7 +568,7 @@ public class MetricsCollectionAndEvaluationTest {
var source = new JobVertexID();
var topology = new JobTopology(new VertexInfo(source, Map.of(), 10, 720));
- metricsCollector = new TestingMetricsCollector(topology);
+ metricsCollector = new TestingMetricsCollector(topology, stateStore);
metricsCollector.setJobUpdateTs(startTime);
metricsCollector.updateMetrics(
source,
@@ -588,36 +586,32 @@ public class MetricsCollectionAndEvaluationTest {
// Until window is full (time=200) we keep returning stabilizing metrics
metricsCollector.setClock(Clock.fixed(Instant.ofEpochMilli(50), ZoneId.systemDefault()));
- assertEquals(
- 1, metricsCollector.updateMetrics(context, stateStore).getMetricHistory().size());
+ assertEquals(1, collect().getMetricHistory().size());
assertEquals(1, stateStore.getCollectedMetrics(context).size());
metricsCollector.setClock(Clock.fixed(Instant.ofEpochMilli(60), ZoneId.systemDefault()));
- assertEquals(
- 2, metricsCollector.updateMetrics(context, stateStore).getMetricHistory().size());
+ assertEquals(2, collect().getMetricHistory().size());
assertEquals(2, stateStore.getCollectedMetrics(context).size());
testTolerateMetricsMissingDuringStabilizationPhase(topology);
metricsCollector.setClock(Clock.fixed(Instant.ofEpochMilli(150), ZoneId.systemDefault()));
- assertEquals(
- 3, metricsCollector.updateMetrics(context, stateStore).getMetricHistory().size());
+ assertEquals(3, collect().getMetricHistory().size());
assertEquals(3, stateStore.getCollectedMetrics(context).size());
metricsCollector.setClock(Clock.fixed(Instant.ofEpochMilli(180), ZoneId.systemDefault()));
- assertEquals(
- 4, metricsCollector.updateMetrics(context, stateStore).getMetricHistory().size());
+ assertEquals(4, collect().getMetricHistory().size());
assertEquals(4, stateStore.getCollectedMetrics(context).size());
// Once we reach full time we trim the stabilization metrics
metricsCollector.setClock(Clock.fixed(Instant.ofEpochMilli(260), ZoneId.systemDefault()));
- assertEquals(
- 2, metricsCollector.updateMetrics(context, stateStore).getMetricHistory().size());
+ assertEquals(2, collect().getMetricHistory().size());
assertEquals(2, stateStore.getCollectedMetrics(context).size());
}
private void testTolerateMetricsMissingDuringStabilizationPhase(JobTopology topology) {
var collectorWithMissingMetrics =
- new TestingMetricsCollector<JobID, JobAutoScalerContext<JobID>>(topology) {
+ new TestingMetricsCollector<JobID, JobAutoScalerContext<JobID>>(
+ topology, stateStore) {
@Override
protected Map<JobVertexID, Map<String, FlinkMetric>> queryFilteredMetricNames(
JobAutoScalerContext<JobID> ctx, JobTopology topology) {
@@ -634,9 +628,7 @@ public class MetricsCollectionAndEvaluationTest {
() -> stateStore.getCollectedMetrics(context).size();
int numCollectedMetricsBeforeTest = numCollectedMetricsSupplier.get();
- assertThrows(
- NotReadyException.class,
- () -> collectorWithMissingMetrics.updateMetrics(context, stateStore));
+ assertThrows(NotReadyException.class, () -> collectorWithMissingMetrics.collect(context));
assertEquals(numCollectedMetricsBeforeTest, numCollectedMetricsSupplier.get());
}
@@ -644,7 +636,7 @@ public class MetricsCollectionAndEvaluationTest {
public void testScaleDownWithZeroProcessingRate() throws Exception {
var topology = new JobTopology(new VertexInfo(source1, Map.of(), 2, 720));
- metricsCollector = new TestingMetricsCollector<>(topology);
+ metricsCollector = new TestingMetricsCollector<>(topology, stateStore);
metricsCollector.setJobUpdateTs(startTime);
var conf = context.getConfiguration();
@@ -661,7 +653,8 @@ public class MetricsCollectionAndEvaluationTest {
var collectedMetrics = collectMetrics();
var evaluation =
- evaluator.evaluate(context.getConfiguration(), collectedMetrics, restartTime);
+ evaluator.computeEvaluatedMetrics(
+ null, context.getConfiguration(), collectedMetrics, restartTime);
assertEquals(
0,
evaluation
@@ -712,7 +705,9 @@ public class MetricsCollectionAndEvaluationTest {
.getMetricHistory()
.put(Instant.ofEpochSecond(1234), new CollectedMetrics(newMetrics, Map.of()));
- evaluation = evaluator.evaluate(context.getConfiguration(), collectedMetrics, restartTime);
+ evaluation =
+ evaluator.computeEvaluatedMetrics(
+ null, context.getConfiguration(), collectedMetrics, restartTime);
assertEquals(
3.,
evaluation
@@ -729,16 +724,22 @@ public class MetricsCollectionAndEvaluationTest {
metricsCollector.setClock(clock);
- var collectedMetrics = metricsCollector.updateMetrics(context, stateStore);
+ var collectedMetrics = collect();
assertEquals(1, collectedMetrics.getMetricHistory().size());
assertFalse(collectedMetrics.isFullyCollected());
metricsCollector.setClock(Clock.offset(clock, Duration.ofSeconds(2)));
- collectedMetrics = metricsCollector.updateMetrics(context, stateStore);
+ collectedMetrics = collect();
assertEquals(2, collectedMetrics.getMetricHistory().size());
assertTrue(collectedMetrics.isFullyCollected());
return collectedMetrics;
}
+
+ private CollectedMetricHistory collect() throws Exception {
+ context.getScalingCycleState().setNow(null);
+ metricsCollector.collect(context);
+ return context.getScalingCycleState().getCollectedMetrics();
+ }
}
diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/RecommendedParallelismTest.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/RecommendedParallelismTest.java
index 99cdc8ead..3d9529487 100644
--- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/RecommendedParallelismTest.java
+++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/RecommendedParallelismTest.java
@@ -77,7 +77,8 @@ public class RecommendedParallelismTest {
new TestingMetricsCollector<>(
new JobTopology(
new VertexInfo(source, Map.of(), 1, 720),
- new VertexInfo(sink, Map.of(source, REBALANCE), 1, 720)));
+ new VertexInfo(sink, Map.of(source, REBALANCE), 1, 720)),
+ stateStore);
var defaultConf = context.getConfiguration();
defaultConf.set(AutoScalerOptions.AUTOSCALER_ENABLED, true);
@@ -95,7 +96,7 @@ public class RecommendedParallelismTest {
autoscaler =
new JobAutoScalerImpl<>(
metricsCollector,
- new ScalingMetricEvaluator(List.of()),
+ new ScalingMetricEvaluator<>(List.of()),
new ScalingExecutor<>(eventCollector, stateStore),
eventCollector,
new TestingScalingRealizer<>(),
@@ -112,7 +113,7 @@ public class RecommendedParallelismTest {
context.getConfiguration().setString(AutoScalerOptions.SCALING_ENABLED.key(), "false");
// initially the last evaluated metrics are empty
- assertNull(autoscaler.lastEvaluatedMetrics.get(context.getJobKey()));
+ assertNull(autoscaler.metricsEvaluator.lastEvaluatedMetrics.get(context.getJobKey()));
var now = Instant.ofEpochMilli(0);
setClocksTo(now);
@@ -211,7 +212,7 @@ public class RecommendedParallelismTest {
// after restart while the job is not running the evaluated metrics are gone
autoscaler.scale(context);
assertCollectedMetricsSize(4);
- assertNull(autoscaler.lastEvaluatedMetrics.get(context.getJobKey()));
+ assertNull(autoscaler.metricsEvaluator.lastEvaluatedMetrics.get(context.getJobKey()));
scaledParallelism = ScalingExecutorTest.getScaledParallelism(stateStore, context);
assertEquals(4, scaledParallelism.get(source));
assertEquals(4, scaledParallelism.get(sink));
@@ -246,6 +247,7 @@ public class RecommendedParallelismTest {
private Double getCurrentMetricValue(JobVertexID jobVertexID, ScalingMetric scalingMetric) {
var metric =
autoscaler
+ .metricsEvaluator
.lastEvaluatedMetrics
.get(context.getJobKey())
.getVertexMetrics()
diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/RestApiMetricsCollectorTest.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/RestApiMetricsCollectorTest.java
index 8ac0f8f0c..286416ec1 100644
--- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/RestApiMetricsCollectorTest.java
+++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/RestApiMetricsCollectorTest.java
@@ -20,6 +20,7 @@ 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.FlinkMetric;
+import org.apache.flink.autoscaler.state.InMemoryAutoScalerStateStore;
import org.apache.flink.client.program.rest.RestClusterClient;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
@@ -68,7 +69,9 @@ class RestApiMetricsCollectorTest {
@Test
void testAggregateMultiplePendingRecordsMetricsPerSource() throws Exception {
- var collector = new RestApiMetricsCollector<JobID, JobAutoScalerContext<JobID>>();
+ var collector =
+ new RestApiMetricsCollector<JobID, JobAutoScalerContext<JobID>>(
+ new InMemoryAutoScalerStateStore<>());
JobVertexID jobVertexID = new JobVertexID();
var flinkMetrics =
@@ -163,7 +166,9 @@ class RestApiMetricsCollectorTest {
(c, e) ->
new StandaloneClientHAServices(
miniCluster.getRestAddress().get().toString()));
- var collector = new RestApiMetricsCollector<>();
+ var collector =
+ new RestApiMetricsCollector<>(
+ new InMemoryAutoScalerStateStore<JobID, JobAutoScalerContext<JobID>>());
// Metrics might not be available yet, and the JobManager exposes the slot metrics
// before the TaskManager has registered its slots (reporting 0). Retry the assertion
// until the slots are registered, or the timeout is reached.
@@ -245,7 +250,9 @@ class RestApiMetricsCollectorTest {
conf,
new UnregisteredMetricsGroup(),
() -> client);
- var collector = new RestApiMetricsCollector<JobID, JobAutoScalerContext<JobID>>();
+ var collector =
+ new RestApiMetricsCollector<JobID, JobAutoScalerContext<JobID>>(
+ new InMemoryAutoScalerStateStore<>());
assertThrows(RuntimeException.class, () -> collector.queryTmMetrics(context));
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 326f24dd9..6213104c5 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
@@ -20,6 +20,7 @@ package org.apache.flink.autoscaler;
import org.apache.flink.api.common.JobID;
import org.apache.flink.autoscaler.config.AutoScalerOptions;
import org.apache.flink.autoscaler.event.TestingEventCollector;
+import org.apache.flink.autoscaler.metrics.CollectedMetricHistory;
import org.apache.flink.autoscaler.metrics.EvaluatedMetrics;
import org.apache.flink.autoscaler.metrics.EvaluatedScalingMetric;
import org.apache.flink.autoscaler.metrics.ScalingMetric;
@@ -52,6 +53,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.TreeMap;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -1094,7 +1096,7 @@ public class ScalingExecutorTest {
*/
@SafeVarargs
private ScalingExecutor<JobID, JobAutoScalerContext<JobID>> executorWith(
- ScalingExecutorPlugin<JobID, JobAutoScalerContext<JobID>>... plugins) {
+ ScalingExecutorPlugin<JobID>... plugins) {
var instances = new ArrayList<String>();
for (var p : plugins) {
var name = p.getClass().getName();
@@ -1109,8 +1111,7 @@ public class ScalingExecutorTest {
@Test
void testFilterApprovesScaling() throws Exception {
// A filter that approves scaling (passes through unchanged).
- ScalingExecutorPlugin<JobID, JobAutoScalerContext<JobID>> approveFilter =
- (ctx, summaries) -> summaries;
+ ScalingExecutorPlugin<JobID> approveFilter = (ctx, summaries) -> summaries;
var executorWithFilter = executorWith(approveFilter);
@@ -1131,8 +1132,7 @@ public class ScalingExecutorTest {
@Test
void testFilterVetoesScaling() throws Exception {
// A filter that vetoes scaling.
- ScalingExecutorPlugin<JobID, JobAutoScalerContext<JobID>> vetoFilter =
- (ctx, summaries) -> Collections.emptyMap();
+ ScalingExecutorPlugin<JobID> vetoFilter = (ctx, summaries) -> Collections.emptyMap();
var executorWithFilter = executorWith(vetoFilter);
@@ -1153,7 +1153,7 @@ public class ScalingExecutorTest {
@Test
void testFilterModifiesSummaries() throws Exception {
// A filter that removes one vertex from scaling summaries.
- ScalingExecutorPlugin<JobID, JobAutoScalerContext<JobID>> modifyFilter =
+ ScalingExecutorPlugin<JobID> modifyFilter =
(ctx, summaries) -> {
// Keep only the first entry
var firstEntry = summaries.entrySet().iterator().next();
@@ -1182,7 +1182,7 @@ public class ScalingExecutorTest {
@Test
void testFilterReturnsEmptyMapVetoesScaling() throws Exception {
// A filter that returns empty map (no vertices left to scale).
- ScalingExecutorPlugin<JobID, JobAutoScalerContext<JobID>> emptyMapFilter =
+ ScalingExecutorPlugin<JobID> emptyMapFilter =
(ctx, summaries) -> Collections.emptyMap();
var executorWithFilter = executorWith(emptyMapFilter);
@@ -1204,12 +1204,10 @@ public class ScalingExecutorTest {
@Test
void testMultipleFiltersChained() throws Exception {
// First filter: approves scaling
- ScalingExecutorPlugin<JobID, JobAutoScalerContext<JobID>> approveFilter =
- (ctx, summaries) -> summaries;
+ ScalingExecutorPlugin<JobID> approveFilter = (ctx, summaries) -> summaries;
// Second filter: vetoes scaling
- ScalingExecutorPlugin<JobID, JobAutoScalerContext<JobID>> vetoFilter =
- (ctx, summaries) -> Collections.emptyMap();
+ ScalingExecutorPlugin<JobID> vetoFilter = (ctx, summaries) -> Collections.emptyMap();
var executorWithFilters = executorWith(approveFilter, vetoFilter);
@@ -1230,10 +1228,8 @@ public class ScalingExecutorTest {
@Test
void testMultipleFiltersAllApprove() throws Exception {
// Both filters approve
- ScalingExecutorPlugin<JobID, JobAutoScalerContext<JobID>> filter1 =
- (ctx, summaries) -> summaries;
- ScalingExecutorPlugin<JobID, JobAutoScalerContext<JobID>> filter2 =
- (ctx, summaries) -> summaries;
+ ScalingExecutorPlugin<JobID> filter1 = (ctx, summaries) -> summaries;
+ ScalingExecutorPlugin<JobID> filter2 = (ctx, summaries) -> summaries;
var executorWithFilters = executorWith(filter1, filter2);
@@ -1281,12 +1277,11 @@ public class ScalingExecutorTest {
// A filter that asserts it receives the expected context parameters
var receivedContextHolder =
new Object() {
- ScalingExecutorPlugin.Context<JobID, JobAutoScalerContext<JobID>>
- pluginContext;
+ ScalingExecutorPlugin.Context<JobID> pluginContext;
Map<JobVertexID, ScalingSummary> summaries;
};
- ScalingExecutorPlugin<JobID, JobAutoScalerContext<JobID>> captureFilter =
+ ScalingExecutorPlugin<JobID> captureFilter =
(ctx, summaries) -> {
receivedContextHolder.pluginContext = ctx;
receivedContextHolder.summaries = summaries;
@@ -1296,6 +1291,10 @@ public class ScalingExecutorTest {
var executorWithFilter = executorWith(captureFilter);
var now = Instant.now();
+ context.getScalingCycleState().setEvaluatedMetrics(metrics);
+ context.getScalingCycleState()
+ .setCollectedMetrics(
+ new CollectedMetricHistory(jobTopology, new TreeMap<>(), now));
executorWithFilter.scaleResource(
context,
metrics,
@@ -1306,11 +1305,19 @@ public class ScalingExecutorTest {
new DelayedScaleDown());
assertThat(receivedContextHolder.pluginContext).isNotNull();
- assertThat(receivedContextHolder.pluginContext.getAutoScalerContext())
- .isSameAs(context);
+ assertThat(receivedContextHolder.pluginContext.getJobKey())
+ .isEqualTo(context.getJobKey());
assertThat(receivedContextHolder.pluginContext.getConfiguration()).isNotNull();
assertThat(receivedContextHolder.pluginContext.getEvaluatedMetrics()).isSameAs(metrics);
assertThat(receivedContextHolder.pluginContext.getJobTopology()).isSameAs(jobTopology);
+ // The plugin context is a JobAutoScalerContext sharing the canonical cycle state and
+ // metric group.
+ assertThat(receivedContextHolder.pluginContext)
+ .isInstanceOf(JobAutoScalerContext.class);
+ assertThat(receivedContextHolder.pluginContext.getScalingCycleState())
+ .isSameAs(context.getScalingCycleState());
+ assertThat(receivedContextHolder.pluginContext.getMetricGroup())
+ .isSameAs(context.getMetricGroup());
assertThat(receivedContextHolder.summaries).isNotEmpty();
// Both vertices should be in the summaries since both are above target
assertThat(receivedContextHolder.summaries).containsKey(source);
@@ -1323,7 +1330,7 @@ public class ScalingExecutorTest {
var executionOrder = new ArrayList<String>();
// High priority filter (priority = -10, should execute first)
- ScalingExecutorPlugin<JobID, JobAutoScalerContext<JobID>> highPriorityFilter =
+ ScalingExecutorPlugin<JobID> highPriorityFilter =
new ScalingExecutorPlugin<>() {
@Override
public int priority() {
@@ -1332,8 +1339,7 @@ public class ScalingExecutorTest {
@Override
public Map<JobVertexID, ScalingSummary> apply(
- ScalingExecutorPlugin.Context<JobID, JobAutoScalerContext<JobID>>
- ctx,
+ ScalingExecutorPlugin.Context<JobID> ctx,
Map<JobVertexID, ScalingSummary> summaries) {
executionOrder.add("high");
return summaries;
@@ -1341,12 +1347,11 @@ public class ScalingExecutorTest {
};
// Default priority filter (priority = 0, should execute second)
- ScalingExecutorPlugin<JobID, JobAutoScalerContext<JobID>> defaultPriorityFilter =
+ ScalingExecutorPlugin<JobID> defaultPriorityFilter =
new ScalingExecutorPlugin<>() {
@Override
public Map<JobVertexID, ScalingSummary> apply(
- ScalingExecutorPlugin.Context<JobID, JobAutoScalerContext<JobID>>
- ctx,
+ ScalingExecutorPlugin.Context<JobID> ctx,
Map<JobVertexID, ScalingSummary> summaries) {
executionOrder.add("default");
return summaries;
@@ -1354,7 +1359,7 @@ public class ScalingExecutorTest {
};
// Low priority filter (priority = 10, should execute last)
- ScalingExecutorPlugin<JobID, JobAutoScalerContext<JobID>> lowPriorityFilter =
+ ScalingExecutorPlugin<JobID> lowPriorityFilter =
new ScalingExecutorPlugin<>() {
@Override
public int priority() {
@@ -1363,8 +1368,7 @@ public class ScalingExecutorTest {
@Override
public Map<JobVertexID, ScalingSummary> apply(
- ScalingExecutorPlugin.Context<JobID, JobAutoScalerContext<JobID>>
- ctx,
+ ScalingExecutorPlugin.Context<JobID> ctx,
Map<JobVertexID, ScalingSummary> summaries) {
executionOrder.add("low");
return summaries;
@@ -1394,8 +1398,7 @@ public class ScalingExecutorTest {
@Test
void testApprovingPluginAllowsScalingWithoutVetoEvent() throws Exception {
- ScalingExecutorPlugin<JobID, JobAutoScalerContext<JobID>> approveFilter =
- (ctx, summaries) -> summaries;
+ ScalingExecutorPlugin<JobID> approveFilter = (ctx, summaries) -> summaries;
var executorWithFilter = executorWith(approveFilter);
@@ -1414,8 +1417,7 @@ public class ScalingExecutorTest {
@Test
void testVetoEventEmittedWhenPluginVetoes() throws Exception {
- ScalingExecutorPlugin<JobID, JobAutoScalerContext<JobID>> vetoFilter =
- (ctx, summaries) -> Collections.emptyMap();
+ ScalingExecutorPlugin<JobID> vetoFilter = (ctx, summaries) -> Collections.emptyMap();
var executorWithFilter = executorWith(vetoFilter);
@@ -1440,10 +1442,8 @@ public class ScalingExecutorTest {
@Test
void testVetoAfterApproveEmitsVetoEvent() throws Exception {
- ScalingExecutorPlugin<JobID, JobAutoScalerContext<JobID>> approveFilter =
- (ctx, summaries) -> summaries;
- ScalingExecutorPlugin<JobID, JobAutoScalerContext<JobID>> vetoFilter =
- (ctx, summaries) -> Collections.emptyMap();
+ ScalingExecutorPlugin<JobID> approveFilter = (ctx, summaries) -> summaries;
+ ScalingExecutorPlugin<JobID> vetoFilter = (ctx, summaries) -> Collections.emptyMap();
var executorWithFilters = executorWith(approveFilter, vetoFilter);
@@ -1486,8 +1486,7 @@ public class ScalingExecutorTest {
@Test
void testLegacyFallbackClassKeyResolvesPlugin() throws Exception {
- ScalingExecutorPlugin<JobID, JobAutoScalerContext<JobID>> vetoFilter =
- (ctx, summaries) -> Collections.emptyMap();
+ ScalingExecutorPlugin<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);
@@ -1521,8 +1520,7 @@ public class ScalingExecutorTest {
// 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();
+ ScalingExecutorPlugin<JobID> vetoFilter = (ctx, summaries) -> Collections.emptyMap();
var name = vetoFilter.getClass().getName();
// Set ONLY the legacy form of 'scaling.custom-executors'.
conf.setString(
@@ -1564,7 +1562,7 @@ public class ScalingExecutorTest {
var captured =
new java.util.concurrent.atomic.AtomicReference<
org.apache.flink.configuration.Configuration>();
- ScalingExecutorPlugin<JobID, JobAutoScalerContext<JobID>> capturingPlugin =
+ ScalingExecutorPlugin<JobID> capturingPlugin =
(ctx, summaries) -> {
captured.set(ctx.getConfiguration());
return summaries;
@@ -1610,7 +1608,7 @@ public class ScalingExecutorTest {
var captured =
new java.util.concurrent.atomic.AtomicReference<
org.apache.flink.configuration.Configuration>();
- ScalingExecutorPlugin<JobID, JobAutoScalerContext<JobID>> capturingPlugin =
+ ScalingExecutorPlugin<JobID> capturingPlugin =
(ctx, summaries) -> {
captured.set(ctx.getConfiguration());
return summaries;
@@ -1646,7 +1644,7 @@ public class ScalingExecutorTest {
var captured =
new java.util.concurrent.atomic.AtomicReference<
org.apache.flink.configuration.Configuration>();
- ScalingExecutorPlugin<JobID, JobAutoScalerContext<JobID>> capturingPlugin =
+ ScalingExecutorPlugin<JobID> capturingPlugin =
(ctx, summaries) -> {
captured.set(ctx.getConfiguration());
return summaries;
diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingMetricCollectorTest.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingMetricCollectorTest.java
index 70e3f539f..bb2150460 100644
--- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingMetricCollectorTest.java
+++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingMetricCollectorTest.java
@@ -22,6 +22,7 @@ import org.apache.flink.api.common.JobStatus;
import org.apache.flink.autoscaler.exceptions.NotReadyException;
import org.apache.flink.autoscaler.metrics.FlinkMetric;
import org.apache.flink.autoscaler.metrics.MetricNotFoundException;
+import org.apache.flink.autoscaler.state.InMemoryAutoScalerStateStore;
import org.apache.flink.autoscaler.topology.IOMetrics;
import org.apache.flink.autoscaler.topology.JobTopology;
import org.apache.flink.autoscaler.topology.VertexInfo;
@@ -215,7 +216,9 @@ public class ScalingMetricCollectorTest {
+ "}";
JobDetailsInfo jobDetailsInfo = new ObjectMapper().readValue(s, JobDetailsInfo.class);
- var metricsCollector = new RestApiMetricsCollector<>();
+ var metricsCollector =
+ new RestApiMetricsCollector<>(
+ new InMemoryAutoScalerStateStore<JobID, JobAutoScalerContext<JobID>>());
var sourceId = JobVertexID.fromHexString("bc764cd8ddf7a0cff126f51c16239658");
var sinkId = JobVertexID.fromHexString("20ba6b65f97481d5570070de90e4e791");
@@ -349,7 +352,9 @@ public class ScalingMetricCollectorTest {
+ "}";
JobDetailsInfo jobDetailsInfo = new ObjectMapper().readValue(s, JobDetailsInfo.class);
- var metricsCollector = new RestApiMetricsCollector<>();
+ var metricsCollector =
+ new RestApiMetricsCollector<>(
+ new InMemoryAutoScalerStateStore<JobID, JobAutoScalerContext<JobID>>());
assertThrows(
NotReadyException.class, () -> metricsCollector.getJobTopology(jobDetailsInfo));
}
@@ -360,7 +365,9 @@ public class ScalingMetricCollectorTest {
"{\"jid\":\"a1b1b53c7c71e7199aa8c43bc703fe7f\",\"name\":\"basic-example\",\"isStoppable\":false,\"state\":\"RUNNING\",\"start-time\":1697114719143,\"end-time\":-1,\"duration\":60731,\"maxParallelism\":-1,\"now\":1697114779874,\"timestamps\":{\"CANCELLING\":0,\"INITIALIZING\":1697114719143,\"RUNNING\":1697114719743,\"CANCELED\":0,\"FINISHED\":0,\"FAILED\":0,\"RESTARTING\":0,\"FAILING\":0,\"CREATED\":1697114719343,\"SUSPENDED\":0,\"RECONCILING\":0},\"vertices\":[{\"id\":\"b [...]
JobDetailsInfo jobDetailsInfo = new ObjectMapper().readValue(s, JobDetailsInfo.class);
- var metricsCollector = new RestApiMetricsCollector();
+ var metricsCollector =
+ new RestApiMetricsCollector(
+ new InMemoryAutoScalerStateStore<JobID, JobAutoScalerContext<JobID>>());
metricsCollector.getJobTopology(jobDetailsInfo);
}
@@ -381,7 +388,9 @@ public class ScalingMetricCollectorTest {
List.of(),
Map.of(),
new JobPlanInfo.RawJson(""));
- var metricsCollector = new RestApiMetricsCollector<>();
+ var metricsCollector =
+ new RestApiMetricsCollector<>(
+ new InMemoryAutoScalerStateStore<JobID, JobAutoScalerContext<JobID>>());
assertEquals(Instant.ofEpochMilli(3L), metricsCollector.getJobRunningTs(details));
}
@@ -389,7 +398,8 @@ public class ScalingMetricCollectorTest {
public void testQueryNamesOnTopologyChange() {
var metricNameQueryCounter = new HashMap<JobVertexID, Integer>();
var collector =
- new RestApiMetricsCollector<JobID, JobAutoScalerContext<JobID>>() {
+ new RestApiMetricsCollector<JobID, JobAutoScalerContext<JobID>>(
+ new InMemoryAutoScalerStateStore<>()) {
@Override
protected Map<String, FlinkMetric> getFilteredVertexMetricNames(
RestClusterClient<?> rc, JobID id, JobVertexID jvi, JobTopology t) {
@@ -453,7 +463,8 @@ public class ScalingMetricCollectorTest {
public void testRequiredMetrics() {
List<String> metricList = new ArrayList<>();
RestApiMetricsCollector<JobID, JobAutoScalerContext<JobID>> testCollector =
- new RestApiMetricsCollector<>() {
+ new RestApiMetricsCollector<>(
+ new InMemoryAutoScalerStateStore<JobID, JobAutoScalerContext<JobID>>()) {
@Override
protected Collection<String> queryAggregatedMetricNames(
RestClusterClient<?> restClient, JobID jobID, JobVertexID jobVertexID) {
@@ -475,7 +486,9 @@ public class ScalingMetricCollectorTest {
@Test
public void testQueryMetricNames() throws Exception {
- var testCollector = new RestApiMetricsCollector<JobID, JobAutoScalerContext<JobID>>();
+ var testCollector =
+ new RestApiMetricsCollector<JobID, JobAutoScalerContext<JobID>>(
+ new InMemoryAutoScalerStateStore<>());
var response = new ArrayList<AggregatedMetric>();
var client =
new RestClusterClient<>(
@@ -545,7 +558,10 @@ public class ScalingMetricCollectorTest {
new VertexInfo(source, Map.of(), 1, 1),
new VertexInfo(sink, Map.of(source, REBALANCE), 1, 1));
- var metricCollector = new TestingMetricsCollector<>(topology);
+ var metricCollector =
+ new TestingMetricsCollector<>(
+ topology,
+ new InMemoryAutoScalerStateStore<JobID, JobAutoScalerContext<JobID>>());
assertThrows(
MetricNotFoundException.class,
diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingMetricEvaluatorTest.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingMetricEvaluatorTest.java
index d914337f2..e979ce4bf 100644
--- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingMetricEvaluatorTest.java
+++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingMetricEvaluatorTest.java
@@ -17,6 +17,7 @@
package org.apache.flink.autoscaler;
+import org.apache.flink.api.common.JobID;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.autoscaler.config.AutoScalerOptions;
import org.apache.flink.autoscaler.metrics.CollectedMetricHistory;
@@ -45,6 +46,7 @@ import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
+import static org.apache.flink.autoscaler.TestingAutoscalerUtils.createDefaultJobAutoScalerContext;
import static org.apache.flink.autoscaler.config.AutoScalerOptions.CATCH_UP_DURATION;
import static org.apache.flink.autoscaler.config.AutoScalerOptions.PREFER_TRACKED_RESTART_TIME;
import static org.apache.flink.autoscaler.config.AutoScalerOptions.RESTART_TIME;
@@ -55,12 +57,14 @@ import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertTrue;
/** Scaling evaluator test. */
public class ScalingMetricEvaluatorTest {
- private ScalingMetricEvaluator evaluator = new ScalingMetricEvaluator(List.of());
+ private ScalingMetricEvaluator<JobID, JobAutoScalerContext<JobID>> evaluator =
+ new ScalingMetricEvaluator<>(List.of());
@Test
public void testLagBasedSourceScaling() {
@@ -119,7 +123,8 @@ public class ScalingMetricEvaluatorTest {
conf.set(CATCH_UP_DURATION, Duration.ofSeconds(2));
var evaluatedMetrics =
evaluator
- .evaluate(
+ .computeEvaluatedMetrics(
+ null,
conf,
new CollectedMetricHistory(topology, metricHistory, Instant.now()),
Duration.ZERO)
@@ -153,7 +158,8 @@ public class ScalingMetricEvaluatorTest {
conf.set(CATCH_UP_DURATION, Duration.ofSeconds(1));
evaluatedMetrics =
evaluator
- .evaluate(
+ .computeEvaluatedMetrics(
+ null,
conf,
new CollectedMetricHistory(topology, metricHistory, Instant.now()),
Duration.ZERO)
@@ -176,7 +182,8 @@ public class ScalingMetricEvaluatorTest {
evaluatedMetrics =
evaluator
- .evaluate(
+ .computeEvaluatedMetrics(
+ null,
conf,
new CollectedMetricHistory(topology, metricHistory, Instant.now()),
Duration.ZERO)
@@ -198,7 +205,8 @@ public class ScalingMetricEvaluatorTest {
conf.set(CATCH_UP_DURATION, Duration.ZERO);
evaluatedMetrics =
evaluator
- .evaluate(
+ .computeEvaluatedMetrics(
+ null,
conf,
new CollectedMetricHistory(topology, metricHistory, Instant.now()),
Duration.ZERO)
@@ -257,7 +265,8 @@ public class ScalingMetricEvaluatorTest {
conf.set(CATCH_UP_DURATION, Duration.ofMinutes(1));
evaluatedMetrics =
evaluator
- .evaluate(
+ .computeEvaluatedMetrics(
+ null,
conf,
new CollectedMetricHistory(topology, metricHistory, Instant.now()),
Duration.ZERO)
@@ -327,14 +336,17 @@ public class ScalingMetricEvaluatorTest {
conf.set(
AutoScalerOptions.customEvaluatorClassOption(customEvaluatorName),
TestCustomEvaluator.class.getName());
- var customEvaluatorAwareEvaluator = new ScalingMetricEvaluator(List.of(customEvaluator));
+ var customEvaluatorAwareEvaluator =
+ new ScalingMetricEvaluator<JobID, JobAutoScalerContext<JobID>>(
+ List.of(customEvaluator));
+
+ var collectedMetrics = new CollectedMetricHistory(topology, metricHistory, Instant.now());
+ var ctx = createDefaultJobAutoScalerContext();
+ ctx.getScalingCycleState().setCollectedMetrics(collectedMetrics);
var evaluatedMetrics =
customEvaluatorAwareEvaluator
- .evaluate(
- conf,
- new CollectedMetricHistory(topology, metricHistory, Instant.now()),
- Duration.ZERO)
+ .computeEvaluatedMetrics(ctx, conf, collectedMetrics, Duration.ZERO)
.getVertexMetrics();
assertEquals(
@@ -363,6 +375,49 @@ public class ScalingMetricEvaluatorTest {
assertFalse(evaluatedMetrics.get(sink).containsKey(ScalingMetric.LAG));
}
+ @Test
+ void testEvaluatorPluginContextExtendsCanonicalContext() {
+ var vertex = new JobVertexID();
+ var topology = new JobTopology(new VertexInfo(vertex, Collections.emptyMap(), 1, 1, null));
+ var metricHistory = new TreeMap<Instant, CollectedMetrics>();
+ metricHistory.put(
+ Instant.ofEpochMilli(1), new CollectedMetrics(Map.of(vertex, Map.of()), Map.of()));
+ var collectedMetrics = new CollectedMetricHistory(topology, metricHistory, Instant.EPOCH);
+
+ var autoScalerContext = createDefaultJobAutoScalerContext();
+ autoScalerContext.getScalingCycleState().setCollectedMetrics(collectedMetrics);
+ autoScalerContext.getScalingCycleState().setRestartTime(Duration.ofSeconds(7));
+ autoScalerContext.getConfiguration().setString("job.key", "job-value");
+
+ var pluginOverrides = new Configuration();
+ pluginOverrides.setString("evaluator.key", "value");
+ Map<JobVertexID, Map<ScalingMetric, EvaluatedScalingMetric>> evaluatedVertexMetrics =
+ Map.of();
+
+ var pluginContext =
+ new ScalingMetricsEvaluatorPlugin.Context<>(
+ autoScalerContext, pluginOverrides, evaluatedVertexMetrics, true);
+
+ // It is a JobAutoScalerContext sharing the canonical context's cycle state and metric
+ // group.
+ assertInstanceOf(JobAutoScalerContext.class, pluginContext);
+ assertSame(autoScalerContext.getScalingCycleState(), pluginContext.getScalingCycleState());
+ assertSame(autoScalerContext.getMetricGroup(), pluginContext.getMetricGroup());
+
+ // getConfiguration() overlays this plugin's overrides on top of the job configuration.
+ assertEquals("job-value", pluginContext.getConfiguration().getString("job.key", null));
+ assertEquals("value", pluginContext.getConfiguration().getString("evaluator.key", null));
+
+ // Cycle-derived accessors read through the shared cycle state.
+ assertSame(topology, pluginContext.getJobTopology());
+ assertEquals(Duration.ofSeconds(7), pluginContext.getRestartTime());
+ assertEquals(metricHistory, pluginContext.getMetricsHistory());
+
+ // Plugin-specific fields.
+ assertSame(evaluatedVertexMetrics, pluginContext.getEvaluatedVertexMetrics());
+ assertTrue(pluginContext.isProcessingBacklog());
+ }
+
@Test
public void testUtilizationBoundaryComputation() {
@@ -1040,16 +1095,19 @@ public class ScalingMetricEvaluatorTest {
ScalingMetricsEvaluatorPlugin customEvaluator = new TestCustomEvaluator();
var evaluatedMetrics = new HashMap<ScalingMetric, EvaluatedScalingMetric>();
+ var ctx = createDefaultJobAutoScalerContext();
+ ctx.getScalingCycleState()
+ .setCollectedMetrics(
+ new CollectedMetricHistory(topology, metricHistory, Instant.EPOCH));
+
var testCustomEvaluationSession =
Tuple2.of(
customEvaluator,
- new ScalingMetricsEvaluatorPlugin.Context(
+ new ScalingMetricsEvaluatorPlugin.Context<>(
+ ctx,
new UnmodifiableConfiguration(conf),
- Collections.unmodifiableSortedMap(metricHistory),
Collections.unmodifiableMap(new HashMap<>()),
- topology,
- false,
- Duration.ZERO));
+ false));
var testCustomEvaluatorResult =
ScalingMetricEvaluator.runCustomEvaluator(
@@ -1142,7 +1200,7 @@ public class ScalingMetricEvaluatorTest {
void testGetCustomEvaluatorIfRequired() {
ScalingMetricsEvaluatorPlugin testCustomEvaluator = new TestCustomEvaluator();
var customEvaluatorAwareEvaluator =
- new ScalingMetricEvaluator(List.of(testCustomEvaluator));
+ new ScalingMetricEvaluator<>(List.of(testCustomEvaluator));
String testCustomEvaluatorName = "test-custom-evaluator";
String testCustomEvaluatorClassName = TestCustomEvaluator.class.getName();
@@ -1227,28 +1285,9 @@ public class ScalingMetricEvaluatorTest {
conf.removeConfig(classOpt);
// Case 7: No custom evaluators registered at all -> null even when configured.
- var noCustomEvaluator = new ScalingMetricEvaluator(Collections.emptyList());
+ var noCustomEvaluator = new ScalingMetricEvaluator<>(Collections.emptyList());
conf.set(AutoScalerOptions.CUSTOM_EVALUATORS, List.of(testCustomEvaluatorName));
conf.set(classOpt, testCustomEvaluatorClassName);
assertNull(noCustomEvaluator.getCustomEvaluatorIfRequired(conf));
}
-
- @Test
- void testMergeEvaluatorConfig() {
- var jobConf = new Configuration();
- jobConf.setString("only-job", "j");
- jobConf.setString("shared", "job-value");
-
- var evaluatorConf = new Configuration();
- evaluatorConf.setString("only-evaluator", "e");
- evaluatorConf.setString("shared", "evaluator-value");
-
- var merged = ScalingMetricEvaluator.mergeEvaluatorConfig(jobConf, evaluatorConf);
-
- // Job-only and evaluator-only keys are both visible.
- assertEquals("j", merged.getString("only-job", null));
- assertEquals("e", merged.getString("only-evaluator", null));
- // On a key present in both, the evaluator-specific value wins.
- assertEquals("evaluator-value", merged.getString("shared", null));
- }
}
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
index c121f01fe..2f3de65e0 100644
--- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/TestScalingExecutor.java
+++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/TestScalingExecutor.java
@@ -48,8 +48,7 @@ import java.util.Map;
* 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> {
+public class TestScalingExecutor<KEY> implements ScalingExecutorPlugin<KEY> {
private static final Logger LOG = LoggerFactory.getLogger(TestScalingExecutor.class);
@@ -71,13 +70,13 @@ public class TestScalingExecutor<KEY, Context extends JobAutoScalerContext<KEY>>
@Override
public Map<JobVertexID, ScalingSummary> apply(
- ScalingExecutorPlugin.Context<KEY, Context> context,
+ ScalingExecutorPlugin.Context<KEY> 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());
+ avgCpuLoad = queryAverageTmCpuLoad(context);
} catch (Exception e) {
LOG.warn(
"Failed to query TaskManager CPU load via REST API. Vetoing scaling as a"
@@ -179,7 +178,7 @@ public class TestScalingExecutor<KEY, Context extends JobAutoScalerContext<KEY>>
*
* @return the average CPU load across all TaskManagers, or 0.0 if not available.
*/
- private double queryAverageTmCpuLoad(Context context) throws Exception {
+ private double queryAverageTmCpuLoad(JobAutoScalerContext<?> context) throws Exception {
try (var restClient = context.getRestClusterClient()) {
var parameters = new AggregateTaskManagerMetricsParameters();
var queryParamIt = parameters.getQueryParameters().iterator();
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
index db5ab7ae0..a19da8608 100644
--- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/TestScalingExecutorTest.java
+++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/TestScalingExecutorTest.java
@@ -19,6 +19,7 @@ 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.CollectedMetricHistory;
import org.apache.flink.autoscaler.metrics.EvaluatedMetrics;
import org.apache.flink.autoscaler.metrics.EvaluatedScalingMetric;
import org.apache.flink.autoscaler.metrics.ScalingMetric;
@@ -43,9 +44,11 @@ import org.apache.flink.runtime.rest.messages.job.metrics.AggregatedTaskManagerM
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import java.time.Instant;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.TreeMap;
import java.util.concurrent.CompletableFuture;
import static org.apache.flink.autoscaler.topology.ShipStrategy.HASH;
@@ -70,16 +73,19 @@ class TestScalingExecutorTest {
new VertexInfo(sink, Map.of(source, HASH), 10, 1000, false, null));
}
- private ScalingExecutorPlugin.Context<JobID, JobAutoScalerContext<JobID>> createPluginContext(
+ private ScalingExecutorPlugin.Context<JobID> createPluginContext(
JobAutoScalerContext<JobID> autoScalerContext) {
- return new ScalingExecutorPlugin.Context<>(
- autoScalerContext, conf, createMetrics(), jobTopology);
+ var cycleState = autoScalerContext.getScalingCycleState();
+ cycleState.setCollectedMetrics(
+ new CollectedMetricHistory(jobTopology, new TreeMap<>(), Instant.EPOCH));
+ cycleState.setEvaluatedMetrics(createMetrics());
+ return new ScalingExecutorPlugin.Context<>(autoScalerContext, conf);
}
@Test
void testCpuThresholdBoundary() {
var context = createContextWithCpuLoad(0.9);
- var plugin = new TestScalingExecutor<JobID, JobAutoScalerContext<JobID>>(0.9);
+ var plugin = new TestScalingExecutor<JobID>(0.9);
var summaries = createSummaries(source, 10, 20, sink, 10, 20);
@@ -92,7 +98,7 @@ class TestScalingExecutorTest {
void testDefaultThreshold() {
// Uses default threshold (0.8), CPU 0.7 -> below
var context = createContextWithCpuLoad(0.7);
- var plugin = new TestScalingExecutor<JobID, JobAutoScalerContext<JobID>>();
+ var plugin = new TestScalingExecutor<JobID>();
var summaries = createSummaries(source, 10, 20, sink, 10, 15);
@@ -105,7 +111,7 @@ class TestScalingExecutorTest {
void testVetoWhenRestClientFails() {
// REST client that throws an exception
var context = createContextWithFailingRestClient();
- var plugin = new TestScalingExecutor<JobID, JobAutoScalerContext<JobID>>(0.5);
+ var plugin = new TestScalingExecutor<JobID>(0.5);
var summaries = createSummaries(source, 10, 20, sink, 10, 15);
@@ -119,7 +125,7 @@ class TestScalingExecutorTest {
void testVetoWhenAvgCpuBelowThreshold() {
// CPU 0.5 < threshold 0.8
var context = createContextWithCpuLoad(0.5);
- var plugin = new TestScalingExecutor<JobID, JobAutoScalerContext<JobID>>(0.8);
+ var plugin = new TestScalingExecutor<JobID>(0.8);
var summaries = createSummaries(source, 10, 20, sink, 10, 20);
@@ -132,7 +138,7 @@ class TestScalingExecutorTest {
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 plugin = new TestScalingExecutor<JobID>(0.8);
var summaries = createSummaries(source, 10, 20, sink, 10, 15);
@@ -148,7 +154,7 @@ class TestScalingExecutorTest {
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 plugin = new TestScalingExecutor<JobID>(0.5);
var summaries = createSummaries(source, 20, 10, sink, 10, 20);
@@ -162,7 +168,7 @@ class TestScalingExecutorTest {
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 plugin = new TestScalingExecutor<JobID>(0.5);
var summaries = new HashMap<JobVertexID, ScalingSummary>();
summaries.put(sink, new ScalingSummary(10, 20, createVertexMetrics()));
@@ -177,7 +183,7 @@ class TestScalingExecutorTest {
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 plugin = new TestScalingExecutor<JobID>(0.5);
var summaries = createSummaries(source, 10, 30, sink, 10, 15);
@@ -194,7 +200,7 @@ class TestScalingExecutorTest {
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 plugin = new TestScalingExecutor<JobID>(0.5);
var summaries = createSummaries(source, 10, 20, sink, 20, 25);
diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/TestingMetricsCollector.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/TestingMetricsCollector.java
index 07e33b339..1e4f19152 100644
--- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/TestingMetricsCollector.java
+++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/TestingMetricsCollector.java
@@ -20,6 +20,7 @@ package org.apache.flink.autoscaler;
import org.apache.flink.api.common.JobID;
import org.apache.flink.autoscaler.metrics.FlinkMetric;
import org.apache.flink.autoscaler.metrics.TestMetrics;
+import org.apache.flink.autoscaler.state.AutoScalerStateStore;
import org.apache.flink.autoscaler.topology.JobTopology;
import org.apache.flink.client.program.rest.RestClusterClient;
import org.apache.flink.configuration.Configuration;
@@ -64,7 +65,9 @@ public class TestingMetricsCollector<KEY, Context extends JobAutoScalerContext<K
@Setter private Map<JobVertexID, Collection<String>> metricNames = new HashMap<>();
- public TestingMetricsCollector(JobTopology jobTopology) {
+ public TestingMetricsCollector(
+ JobTopology jobTopology, AutoScalerStateStore<KEY, Context> stateStore) {
+ super(stateStore);
this.jobTopology = jobTopology;
}
diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/alignment/AlignmentModeTest.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/alignment/AlignmentModeTest.java
index 316f13296..90cd0f4b1 100644
--- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/alignment/AlignmentModeTest.java
+++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/alignment/AlignmentModeTest.java
@@ -17,6 +17,7 @@
package org.apache.flink.autoscaler.alignment;
+import org.apache.flink.api.common.JobID;
import org.apache.flink.autoscaler.config.AutoScalerOptions;
import org.apache.flink.autoscaler.topology.ShipStrategy;
import org.apache.flink.configuration.Configuration;
@@ -28,6 +29,7 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
+import static org.apache.flink.autoscaler.TestingAutoscalerUtils.createDefaultJobAutoScalerContext;
import static org.assertj.core.api.Assertions.assertThat;
/**
@@ -41,7 +43,7 @@ class AlignmentModeTest {
private final ParallelismAligner.ScalingLimitedEmitter emitter =
(expected, actual) -> emitted.incrementAndGet();
- private ParallelismAlignmentMode.Context ctx(
+ private ParallelismAlignmentMode.Context<JobID> ctx(
int current, int newParallelism, int numSourcePartitions, int maxParallelism) {
return ctx(
current,
@@ -53,7 +55,7 @@ class AlignmentModeTest {
List.of(ShipStrategy.HASH));
}
- private ParallelismAlignmentMode.Context ctx(
+ private ParallelismAlignmentMode.Context<JobID> ctx(
int current,
int newParallelism,
int numSourcePartitions,
@@ -61,7 +63,9 @@ class AlignmentModeTest {
int parallelismLowerLimit,
int parallelismUpperLimit,
Collection<ShipStrategy> inputShipStrategies) {
- return new ParallelismAlignmentMode.Context(
+ return new ParallelismAlignmentMode.Context<>(
+ createDefaultJobAutoScalerContext(),
+ new Configuration(),
null,
current,
newParallelism,
@@ -70,10 +74,7 @@ class AlignmentModeTest {
parallelismLowerLimit,
parallelismUpperLimit,
inputShipStrategies,
- null,
- Map.of(),
- null,
- new Configuration());
+ Map.of());
}
@Test
@@ -95,8 +96,7 @@ class AlignmentModeTest {
void builtInModesFallBackToTargetInsteadOfBlocking() {
// current=22, target=25, N=35 source partitions, upper bound 30: no aligned value preserves
// the scale-up. Built-in modes return the raw target (OFF fallback), and emit no event.
- ParallelismAlignmentMode.Context blocked =
- ctx(22, 25, 35, 128, 20, 30, List.of(ShipStrategy.HASH));
+ var blocked = ctx(22, 25, 35, 128, 20, 30, List.of(ShipStrategy.HASH));
assertThat(BuiltInAlignmentMode.EVENLY_SPREAD.alignParallelism(blocked)).isEqualTo(25);
assertThat(BuiltInAlignmentMode.BALANCED.alignParallelism(blocked)).isEqualTo(25);
assertThat(emitted).hasValue(0);
@@ -108,7 +108,7 @@ class AlignmentModeTest {
// N=12, scale-up to target 7 with the region capped at 8: no divisor in [7, 8]. The new
// built-in modes keep the target (7), whereas the legacy modes relax below the target to
// the nearest divisor 6.
- ParallelismAlignmentMode.Context c = ctx(5, 7, 12, 8);
+ var c = ctx(5, 7, 12, 8);
assertThat(BuiltInAlignmentMode.EVENLY_SPREAD.alignParallelism(c)).isEqualTo(7);
assertThat(KeyGroupOrPartitionsAdjustMode.EVENLY_SPREAD.alignParallelism(c)).isEqualTo(6);
}
@@ -117,8 +117,7 @@ class AlignmentModeTest {
@SuppressWarnings("deprecation")
void legacyModesBlockAndEmit() {
// The same scenario under the deprecated legacy mode blocks (returns current) and emits.
- ParallelismAlignmentMode.Context blocked =
- ctx(22, 25, 35, 128, 20, 30, List.of(ShipStrategy.HASH));
+ var blocked = ctx(22, 25, 35, 128, 20, 30, List.of(ShipStrategy.HASH));
assertThat(KeyGroupOrPartitionsAdjustMode.EVENLY_SPREAD.alignParallelism(blocked, emitter))
.isEqualTo(22);
assertThat(emitted).hasValue(1);
@@ -126,13 +125,12 @@ class AlignmentModeTest {
@Test
void modesDoNotApplyToNonSourceNonHashVertices() {
- ParallelismAlignmentMode.Context rebalance =
- ctx(16, 24, 0, 128, 1, Integer.MAX_VALUE, List.of(ShipStrategy.REBALANCE));
+ var rebalance = ctx(16, 24, 0, 128, 1, Integer.MAX_VALUE, List.of(ShipStrategy.REBALANCE));
assertThat(BuiltInAlignmentMode.BALANCED.isApplicable(rebalance)).isFalse();
assertThat(BuiltInAlignmentMode.EVENLY_SPREAD.isApplicable(rebalance)).isFalse();
assertThat(BuiltInAlignmentMode.OFF.isApplicable(rebalance)).isFalse();
// A source / keyBy vertex is in scope.
- ParallelismAlignmentMode.Context hash = ctx(16, 24, 0, 128);
+ var hash = ctx(16, 24, 0, 128);
assertThat(BuiltInAlignmentMode.BALANCED.isApplicable(hash)).isTrue();
}
diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/alignment/TestAlignmentMode.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/alignment/TestAlignmentMode.java
index e3875b043..61cf3bd0e 100644
--- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/alignment/TestAlignmentMode.java
+++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/alignment/TestAlignmentMode.java
@@ -26,7 +26,7 @@ package org.apache.flink.autoscaler.alignment;
public class TestAlignmentMode implements ParallelismAlignmentMode {
@Override
- public int alignParallelism(Context ctx) {
+ public int alignParallelism(Context<?> ctx) {
int n = ParallelismAligner.numKeyGroupsOrPartitions(ctx);
for (int p = ctx.getNewParallelism(); p >= 1; p--) {
if (n % p == 0) {
diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/config/AutoScalerOptionsTest.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/config/AutoScalerOptionsTest.java
index ce5128f2e..db6beaf04 100644
--- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/config/AutoScalerOptionsTest.java
+++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/config/AutoScalerOptionsTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.autoscaler.config;
import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.Configuration;
import org.junit.jupiter.api.Test;
@@ -53,6 +54,25 @@ public class AutoScalerOptionsTest {
}));
}
+ @Test
+ void testOverlayConfiguration() {
+ var base = new Configuration();
+ base.setString("only-base", "b");
+ base.setString("shared", "base-value");
+
+ var overrides = new Configuration();
+ overrides.setString("only-overrides", "o");
+ overrides.setString("shared", "override-value");
+
+ var merged = AutoScalerOptions.overlayConfiguration(base, overrides);
+
+ // Base-only and override-only keys are both visible.
+ assertThat(merged.getString("only-base", null)).isEqualTo("b");
+ assertThat(merged.getString("only-overrides", null)).isEqualTo("o");
+ // On a key present in both, the override value wins.
+ assertThat(merged.getString("shared", null)).isEqualTo("override-value");
+ }
+
private static List<ConfigOption<?>> retrieveAutoscalerConfigOptions() {
List<ConfigOption<?>> configOptions = new ArrayList<>();
diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/metrics/TestCustomEvaluator.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/metrics/TestCustomEvaluator.java
index 4db951294..0cc5244e3 100644
--- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/metrics/TestCustomEvaluator.java
+++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/metrics/TestCustomEvaluator.java
@@ -34,8 +34,8 @@ public class TestCustomEvaluator implements ScalingMetricsEvaluatorPlugin {
public Map<ScalingMetric, EvaluatedScalingMetric> evaluateVertexMetrics(
JobVertexID vertex,
Map<ScalingMetric, EvaluatedScalingMetric> evaluatedMetrics,
- Context evaluationContext) {
- if (evaluationContext.getTopology().isSource(vertex)) {
+ Context<?> evaluationContext) {
+ if (evaluationContext.getJobTopology().isSource(vertex)) {
var customEvaluatedMetrics = new HashMap<ScalingMetric, EvaluatedScalingMetric>();
customEvaluatedMetrics.put(
ScalingMetric.TARGET_DATA_RATE, EvaluatedScalingMetric.avg(100000.0));
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 0d23f9e0e..68ed968cb 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
@@ -52,14 +52,14 @@ public class AutoscalerFactory {
// Autoscaler SPI integrations sharing the operator's process-wide plugin manager.
Collection<ScalingMetricsEvaluatorPlugin> customEvaluators =
AutoscalerUtils.discoverCustomEvaluators(pluginManager);
- Collection<ScalingExecutorPlugin<ResourceID, KubernetesJobAutoScalerContext>>
- customExecutors = AutoscalerUtils.discoverCustomScalingExecutors(pluginManager);
+ Collection<ScalingExecutorPlugin<ResourceID>> customExecutors =
+ AutoscalerUtils.discoverCustomScalingExecutors(pluginManager);
Collection<ParallelismAlignmentMode> customAlignmentModes =
AutoscalerUtils.discoverCustomAlignmentModes(pluginManager);
return new JobAutoScalerImpl<>(
- new RestApiMetricsCollector<>(),
- new ScalingMetricEvaluator(customEvaluators),
+ new RestApiMetricsCollector<>(stateStore),
+ new ScalingMetricEvaluator<>(customEvaluators),
new ScalingExecutor<>(
eventHandler,
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 6928e0cd3..9ab5a713b 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
@@ -68,9 +68,9 @@ public class AutoscalerUtils {
*/
@SuppressWarnings("unchecked")
public static <KEY, Context extends JobAutoScalerContext<KEY>>
- Collection<ScalingExecutorPlugin<KEY, Context>> discoverCustomScalingExecutors(
+ Collection<ScalingExecutorPlugin<KEY>> discoverCustomScalingExecutors(
PluginManager pluginManager) {
- List<ScalingExecutorPlugin<KEY, Context>> customScalingExecutors = new ArrayList<>();
+ List<ScalingExecutorPlugin<KEY>> customScalingExecutors = new ArrayList<>();
pluginManager
.load(ScalingExecutorPlugin.class)
.forEachRemaining(
@@ -83,7 +83,7 @@ public class AutoscalerUtils {
ConfigConstants.DEFAULT_FLINK_PLUGINS_DIRS),
customScalingExecutor.getClass().getName());
customScalingExecutors.add(
- (ScalingExecutorPlugin<KEY, Context>) customScalingExecutor);
+ (ScalingExecutorPlugin<KEY>) customScalingExecutor);
});
return customScalingExecutors;
}