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 0f66927c [FLINK-40558] Apply in-place scaling only on SCALE spec diff
type
0f66927c is described below
commit 0f66927c42b1e007c12c3e66afd72b00b7556929
Author: Dennis-Mircea Ciupitu <[email protected]>
AuthorDate: Mon Sep 14 17:57:49 2026 +0200
[FLINK-40558] Apply in-place scaling only on SCALE spec diff type
---
.../AbstractFlinkResourceReconciler.java | 9 +--
.../deployment/ApplicationReconcilerTest.java | 66 ++++++++++++++++++++++
2 files changed, 68 insertions(+), 7 deletions(-)
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java
index 3c16b350..3be75396 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java
@@ -154,11 +154,7 @@ public abstract class AbstractFlinkResourceReconciler<
}
triggerSpecChangeEvent(cr, specDiff, ctx.getKubernetesClient());
- // Try scaling if this is not an upgrade/redeploy change
- boolean scaled =
- diffType != DiffType.SAVEPOINT_REDEPLOY
- && diffType != DiffType.UPGRADE
- && scale(ctx, deployConfig);
+ boolean scaled = diffType == DiffType.SCALE && scale(ctx,
deployConfig);
// Reconcile spec change unless scaling was enough
if (scaled || reconcileSpecChange(diffType, ctx, deployConfig,
lastReconciledSpec)) {
@@ -354,8 +350,7 @@ public abstract class AbstractFlinkResourceReconciler<
*
* @param ctx Resource context.
* @param deployConfig Configuration to be deployed.
- * @return True if the scaling is successful
- * @throws Exception
+ * @return True if the in-place scaling is successful
*/
private boolean scale(FlinkResourceContext<CR> ctx, Configuration
deployConfig)
throws Exception {
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java
index a5d87bc9..d1188c7f 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java
@@ -62,6 +62,7 @@ import
org.apache.flink.kubernetes.operator.api.utils.SpecUtils;
import
org.apache.flink.kubernetes.operator.autoscaler.KubernetesJobAutoScalerContext;
import org.apache.flink.kubernetes.operator.config.FlinkOperatorConfiguration;
import
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
+import org.apache.flink.kubernetes.operator.controller.FlinkResourceContext;
import org.apache.flink.kubernetes.operator.exception.UpgradeFailureException;
import org.apache.flink.kubernetes.operator.health.ClusterHealthInfo;
import org.apache.flink.kubernetes.operator.observer.ClusterHealthEvaluator;
@@ -967,6 +968,71 @@ public class ApplicationReconcilerTest extends
OperatorTestBase {
assertFalse(reconStatus.isLastReconciledSpecStable());
}
+ @Test
+ public void testResumedUpgradeDoesNotTriggerInPlaceScaling() throws
Exception {
+ var scaleCounter = new AtomicInteger(0);
+ var cancelCounter = new AtomicInteger(0);
+ var deployCounter = new AtomicInteger(0);
+
+ var nativeService =
+ new NativeFlinkService(
+ kubernetesClient, null, executorService,
operatorConfig, eventRecorder) {
+ @Override
+ public boolean scale(FlinkResourceContext<?> ctx,
Configuration deployConfig) {
+ scaleCounter.incrementAndGet();
+ return true;
+ }
+
+ @Override
+ public CancelResult cancelJob(
+ FlinkDeployment deployment,
+ SuspendMode suspendMode,
+ Configuration conf) {
+ cancelCounter.incrementAndGet();
+
deployment.getStatus().getJobStatus().setState(FINISHED);
+ return CancelResult.completed(null);
+ }
+
+ @Override
+ protected void deployApplicationCluster(JobSpec jobSpec,
Configuration conf) {
+ deployCounter.incrementAndGet();
+ }
+ };
+
+ var ctxFactory =
+ new TestingFlinkResourceContextFactory(
+ configManager, operatorMetricGroup, nativeService,
eventRecorder);
+ FlinkDeployment deployment = TestUtils.buildApplicationCluster();
+ deployment.getMetadata().setGeneration(1L);
+
+ reconciler.reconcile(deployment, context);
+ verifyAndSetRunningJobsToStatus(deployment, flinkService.listJobs());
+
+ // Simulate an upgrade such as the spec diff is IGNORE while the
reconciliation state is
+ // UPGRADING
+
deployment.getStatus().getReconciliationStatus().setState(ReconciliationState.UPGRADING);
+
+ appReconciler.reconcile(ctxFactory.getResourceContext(deployment,
context));
+
+ // The first reconcile starts the upgrade (suspend job) and does not
trigger in-place
+ // scaling
+ assertEquals(0, scaleCounter.get());
+ assertEquals(1, cancelCounter.get());
+ assertEquals(0, deployCounter.get());
+ assertEquals(
+ ReconciliationState.UPGRADING,
+ deployment.getStatus().getReconciliationStatus().getState());
+
+ // The second reconcile completes the upgrade (redeploy job)
+ appReconciler.reconcile(ctxFactory.getResourceContext(deployment,
context));
+ assertEquals(0, scaleCounter.get());
+ assertEquals(1, cancelCounter.get());
+ assertEquals(1, deployCounter.get());
+ assertEquals(
+ ReconciliationState.DEPLOYED,
+ deployment.getStatus().getReconciliationStatus().getState());
+ }
+
@Test
public void testApplyAutoscalerParallelism() throws Exception {
var ctxFactory =
