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 73398a768 [hotfix] Remove redundant autoscaler Plugin discovery (#1198)
73398a768 is described below
commit 73398a7687c84348542715627d8c845e7dc773d0
Author: Dennis-Mircea Ciupitu <[email protected]>
AuthorDate: Wed Sep 16 17:35:09 2026 +0200
[hotfix] Remove redundant autoscaler Plugin discovery (#1198)
---
.../flink/kubernetes/operator/FlinkOperator.java | 7 ++-
.../operator/autoscaler/AutoscalerFactory.java | 23 ++-------
.../operator/autoscaler/AutoscalerPlugins.java | 60 ++++++++++++++++++++++
.../operator/autoscaler/AutoscalerFactoryTest.java | 3 +-
.../TestingFlinkDeploymentController.java | 5 +-
5 files changed, 75 insertions(+), 23 deletions(-)
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java
index cfb7d3ce3..b37aefdfd 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java
@@ -27,6 +27,7 @@ import
org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
import org.apache.flink.kubernetes.operator.api.listener.FlinkResourceListener;
import
org.apache.flink.kubernetes.operator.api.validation.FlinkResourceValidator;
import org.apache.flink.kubernetes.operator.autoscaler.AutoscalerFactory;
+import org.apache.flink.kubernetes.operator.autoscaler.AutoscalerPlugins;
import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
import org.apache.flink.kubernetes.operator.config.FlinkOperatorConfiguration;
import
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
@@ -91,6 +92,7 @@ public class FlinkOperator {
private final EventRecorder eventRecorder;
private final Configuration baseConfig;
private final PluginManager pluginManager;
+ private final AutoscalerPlugins autoscalerPlugins;
public FlinkOperator(@Nullable Configuration conf) {
this(conf, null);
@@ -118,6 +120,7 @@ public class FlinkOperator {
this.operator = createOperator();
this.validators = ValidatorUtils.discoverValidators(configManager,
pluginManager);
this.listeners = ListenerUtils.discoverListeners(configManager,
pluginManager);
+ this.autoscalerPlugins = AutoscalerPlugins.discover(pluginManager);
this.eventRecorder = EventRecorder.create(client, listeners);
this.ctxFactory =
new FlinkResourceContextFactory(configManager, metricGroup,
eventRecorder);
@@ -190,7 +193,7 @@ public class FlinkOperator {
ClusterResourceManager.of(configManager.getDefaultConfig(),
client);
var autoscaler =
AutoscalerFactory.create(
- client, eventRecorder, clusterResourceManager,
pluginManager);
+ client, eventRecorder, clusterResourceManager,
autoscalerPlugins);
var reconcilerFactory = new ReconcilerFactory(eventRecorder,
statusRecorder, autoscaler);
var observerFactory = new
FlinkDeploymentObserverFactory(eventRecorder);
var canaryResourceManager = new
CanaryResourceManager<FlinkDeployment>(configManager);
@@ -214,7 +217,7 @@ public class FlinkOperator {
var metricManager =
MetricManager.createFlinkSessionJobMetricManager(baseConfig,
metricGroup);
var statusRecorder = StatusRecorder.create(client, metricManager,
listeners);
- var autoscaler = AutoscalerFactory.create(client, eventRecorder, null,
pluginManager);
+ var autoscaler = AutoscalerFactory.create(client, eventRecorder, null,
autoscalerPlugins);
var reconciler = new SessionJobReconciler(eventRecorder,
statusRecorder, autoscaler);
var observer = new FlinkSessionJobObserver(eventRecorder);
var canaryResourceManager = new
CanaryResourceManager<FlinkSessionJob>(configManager);
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 68ed968cb..bef98c93f 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/AutoscalerFactory.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/AutoscalerFactory.java
@@ -21,22 +21,15 @@ import org.apache.flink.autoscaler.JobAutoScaler;
import org.apache.flink.autoscaler.JobAutoScalerImpl;
import org.apache.flink.autoscaler.RestApiMetricsCollector;
import org.apache.flink.autoscaler.ScalingExecutor;
-import org.apache.flink.autoscaler.ScalingExecutorPlugin;
import org.apache.flink.autoscaler.ScalingMetricEvaluator;
-import org.apache.flink.autoscaler.alignment.ParallelismAlignmentMode;
-import org.apache.flink.autoscaler.metrics.ScalingMetricsEvaluatorPlugin;
-import org.apache.flink.core.plugin.PluginManager;
import org.apache.flink.kubernetes.operator.autoscaler.state.ConfigMapStore;
import
org.apache.flink.kubernetes.operator.autoscaler.state.KubernetesAutoScalerStateStore;
import org.apache.flink.kubernetes.operator.resources.ClusterResourceManager;
-import org.apache.flink.kubernetes.operator.utils.AutoscalerUtils;
import org.apache.flink.kubernetes.operator.utils.EventRecorder;
import io.fabric8.kubernetes.client.KubernetesClient;
import io.javaoperatorsdk.operator.processing.event.ResourceID;
-import java.util.Collection;
-
/** The factory of {@link JobAutoScaler}. */
public class AutoscalerFactory {
@@ -44,28 +37,20 @@ public class AutoscalerFactory {
KubernetesClient client,
EventRecorder eventRecorder,
ClusterResourceManager clusterResourceManager,
- PluginManager pluginManager) {
+ AutoscalerPlugins plugins) {
var stateStore = new KubernetesAutoScalerStateStore(new
ConfigMapStore(client));
var eventHandler = new KubernetesAutoScalerEventHandler(eventRecorder);
- // Autoscaler SPI integrations sharing the operator's process-wide
plugin manager.
- Collection<ScalingMetricsEvaluatorPlugin> customEvaluators =
- AutoscalerUtils.discoverCustomEvaluators(pluginManager);
- Collection<ScalingExecutorPlugin<ResourceID>> customExecutors =
- AutoscalerUtils.discoverCustomScalingExecutors(pluginManager);
- Collection<ParallelismAlignmentMode> customAlignmentModes =
- AutoscalerUtils.discoverCustomAlignmentModes(pluginManager);
-
return new JobAutoScalerImpl<>(
new RestApiMetricsCollector<>(stateStore),
- new ScalingMetricEvaluator<>(customEvaluators),
+ new ScalingMetricEvaluator<>(plugins.getEvaluators()),
new ScalingExecutor<>(
eventHandler,
stateStore,
clusterResourceManager,
- customExecutors,
- customAlignmentModes),
+ plugins.getScalingExecutors(),
+ plugins.getAlignmentModes()),
eventHandler,
new KubernetesScalingRealizer(),
stateStore);
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/AutoscalerPlugins.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/AutoscalerPlugins.java
new file mode 100644
index 000000000..0f9d90cd8
--- /dev/null
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/autoscaler/AutoscalerPlugins.java
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.operator.autoscaler;
+
+import org.apache.flink.autoscaler.ScalingExecutorPlugin;
+import org.apache.flink.autoscaler.alignment.ParallelismAlignmentMode;
+import org.apache.flink.autoscaler.metrics.ScalingMetricsEvaluatorPlugin;
+import org.apache.flink.core.plugin.PluginManager;
+import org.apache.flink.kubernetes.operator.utils.AutoscalerUtils;
+
+import io.javaoperatorsdk.operator.processing.event.ResourceID;
+import lombok.Getter;
+
+import java.util.Collection;
+
+/** The autoscaler SPI implementations discovered from the operator's plugin
manager. */
+@Getter
+public class AutoscalerPlugins {
+
+ private final Collection<ScalingMetricsEvaluatorPlugin> evaluators;
+ private final Collection<ScalingExecutorPlugin<ResourceID>>
scalingExecutors;
+ private final Collection<ParallelismAlignmentMode> alignmentModes;
+
+ private AutoscalerPlugins(
+ Collection<ScalingMetricsEvaluatorPlugin> evaluators,
+ Collection<ScalingExecutorPlugin<ResourceID>> scalingExecutors,
+ Collection<ParallelismAlignmentMode> alignmentModes) {
+ this.evaluators = evaluators;
+ this.scalingExecutors = scalingExecutors;
+ this.alignmentModes = alignmentModes;
+ }
+
+ /**
+ * Discovers all autoscaler plugins from the given plugin manager.
+ *
+ * @param pluginManager The shared operator plugin manager used for
discovery.
+ * @return The discovered plugins.
+ */
+ public static AutoscalerPlugins discover(PluginManager pluginManager) {
+ return new AutoscalerPlugins(
+ AutoscalerUtils.discoverCustomEvaluators(pluginManager),
+ AutoscalerUtils.discoverCustomScalingExecutors(pluginManager),
+ AutoscalerUtils.discoverCustomAlignmentModes(pluginManager));
+ }
+}
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/AutoscalerFactoryTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/AutoscalerFactoryTest.java
index 64d28e61f..690000199 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/AutoscalerFactoryTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/autoscaler/AutoscalerFactoryTest.java
@@ -50,7 +50,8 @@ public class AutoscalerFactoryTest {
new FlinkResourceEventCollector(),
new FlinkStateSnapshotEventCollector()),
new ClusterResourceManager(Duration.ZERO,
kubernetesClient),
- OperatorPluginUtils.createPluginManager(new
Configuration()));
+ AutoscalerPlugins.discover(
+ OperatorPluginUtils.createPluginManager(new
Configuration())));
Assertions.assertTrue(autoScaler instanceof JobAutoScalerImpl);
}
}
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/TestingFlinkDeploymentController.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/TestingFlinkDeploymentController.java
index 2f2ade454..3a1ed0eff 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/TestingFlinkDeploymentController.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/TestingFlinkDeploymentController.java
@@ -26,6 +26,7 @@ import
org.apache.flink.kubernetes.operator.api.FlinkDeployment;
import org.apache.flink.kubernetes.operator.api.spec.FlinkDeploymentSpec;
import org.apache.flink.kubernetes.operator.api.status.FlinkDeploymentStatus;
import org.apache.flink.kubernetes.operator.autoscaler.AutoscalerFactory;
+import org.apache.flink.kubernetes.operator.autoscaler.AutoscalerPlugins;
import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
import org.apache.flink.kubernetes.operator.health.CanaryResourceManager;
import org.apache.flink.kubernetes.operator.metrics.MetricManager;
@@ -102,7 +103,9 @@ public class TestingFlinkDeploymentController
eventRecorder,
new ClusterResourceManager(
Duration.ZERO,
flinkService.getKubernetesClient()),
- OperatorPluginUtils.createPluginManager(new
Configuration())));
+ AutoscalerPlugins.discover(
+
OperatorPluginUtils.createPluginManager(
+ new Configuration()))));
canaryResourceManager = new CanaryResourceManager<>(configManager);
flinkDeploymentController =
new FlinkDeploymentController(
