[flink-kubernetes-operator] branch release-1.1 updated: [FLINK-28625] Guard against changing target session deployment for sessionjob

[gyfora]

This is an automated email from the ASF dual-hosted git repository.

gyfora pushed a commit to branch release-1.1
in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git


The following commit(s) were added to refs/heads/release-1.1 by this push:
     new 67f0770c [FLINK-28625] Guard against changing target session 
deployment for sessionjob
67f0770c is described below

commit 67f0770c04714aa17fa86ca53b01bac2467a98c5
Author: Aitozi <yuli....@alibaba-inc.com>
AuthorDate: Wed Aug 3 15:32:27 2022 +0800

    [FLINK-28625] Guard against changing target session deployment for 
sessionjob
---
 .../operator/validation/DefaultValidator.java           | 17 +++++++++++++----
 .../operator/validation/DefaultValidatorTest.java       | 11 +++++++++++
 2 files changed, 24 insertions(+), 4 deletions(-)

diff --git 
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java
 
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java
index de3c7b68..93cd0993 100644
--- 
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java
+++ 
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java
@@ -342,6 +342,7 @@ public class DefaultValidator implements 
FlinkResourceValidator {
 
     private Optional<String> validateSessionJobOnly(FlinkSessionJob 
sessionJob) {
         return firstPresent(
+                
validateDeploymentName(sessionJob.getSpec().getDeploymentName()),
                 validateJobNotEmpty(sessionJob),
                 validateNotLastStateUpgradeMode(sessionJob),
                 validateSpecChange(sessionJob),
@@ -400,16 +401,24 @@ public class DefaultValidator implements 
FlinkResourceValidator {
     private Optional<String> validateSpecChange(FlinkSessionJob sessionJob) {
         FlinkSessionJobSpec newSpec = sessionJob.getSpec();
 
-        if (sessionJob.getStatus() == null
-                || sessionJob.getStatus().getReconciliationStatus() == null
-                || 
sessionJob.getStatus().getReconciliationStatus().getLastReconciledSpec()
-                        == null) {
+        if 
(sessionJob.getStatus().getReconciliationStatus().isFirstDeployment()) {
             // New job
             if (newSpec.getJob() != null && 
!newSpec.getJob().getState().equals(JobState.RUNNING)) {
                 return Optional.of("Job must start in running state");
             }
 
             return Optional.empty();
+        } else {
+            var lastReconciledSpec =
+                    sessionJob
+                            .getStatus()
+                            .getReconciliationStatus()
+                            .deserializeLastReconciledSpec();
+            if (!lastReconciledSpec
+                    .getDeploymentName()
+                    .equals(sessionJob.getSpec().getDeploymentName())) {
+                return Optional.of("The deploymentName can't be changed");
+            }
         }
 
         return Optional.empty();
diff --git 
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/validation/DefaultValidatorTest.java
 
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/validation/DefaultValidatorTest.java
index 5189ab47..b6868bfa 100644
--- 
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/validation/DefaultValidatorTest.java
+++ 
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/validation/DefaultValidatorTest.java
@@ -430,6 +430,17 @@ public class DefaultValidatorTest {
                 flinkDeployment -> {},
                 "Invalid session job flinkConfiguration key: 
kubernetes.operator.reconcile.interval."
                         + " Allowed keys are 
[kubernetes.operator.user.artifacts.http.header]");
+
+        testSessionJobValidateWithModifier(
+                sessionJob -> {
+                    sessionJob
+                            .getStatus()
+                            .getReconciliationStatus()
+                            
.serializeAndSetLastReconciledSpec(sessionJob.getSpec(), sessionJob);
+                    
sessionJob.getSpec().setDeploymentName("new-deployment-name");
+                },
+                flinkDeployment -> {},
+                "The deploymentName can't be changed");
     }
 
     private void testSessionJobValidateWithModifier(