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 0cba89cd [FLINK-28625] Guard against changing target session
deployment for sessionjob
0cba89cd is described below
commit 0cba89cdea5e19b92b1000c7e19e325b6d7625a2
Author: Aitozi <[email protected]>
AuthorDate: Wed Aug 3 15:32:27 2022 +0800
[FLINK-28625] Guard against changing target session deployment for
sessionjob
---
.../operator/validation/DefaultValidator.java | 34 ++++++++++------------
.../operator/validation/DefaultValidatorTest.java | 11 +++++++
2 files changed, 27 insertions(+), 18 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 9e92dc93..18fc9e07 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
@@ -89,11 +89,7 @@ public class DefaultValidator implements
FlinkResourceValidator {
deployment.getMetadata().getName(),
deployment.getMetadata().getNamespace()),
validateLogConfig(spec.getLogConfiguration()),
- validateJobSpec(
- spec.getJob(),
- spec.getTaskManager(),
- effectiveConfig,
-
KubernetesDeploymentMode.getDeploymentMode(deployment)),
+ validateJobSpec(spec.getJob(), spec.getTaskManager(),
effectiveConfig),
validateJmSpec(spec.getJobManager(), effectiveConfig),
validateTmSpec(spec.getTaskManager()),
validateSpecChange(deployment, effectiveConfig),
@@ -178,10 +174,7 @@ public class DefaultValidator implements
FlinkResourceValidator {
}
private Optional<String> validateJobSpec(
- JobSpec job,
- @Nullable TaskManagerSpec tm,
- Map<String, String> confMap,
- KubernetesDeploymentMode mode) {
+ JobSpec job, @Nullable TaskManagerSpec tm, Map<String, String>
confMap) {
if (job == null) {
return Optional.empty();
}
@@ -368,6 +361,7 @@ public class DefaultValidator implements
FlinkResourceValidator {
private Optional<String> validateSessionJobOnly(FlinkSessionJob
sessionJob) {
return firstPresent(
+
validateDeploymentName(sessionJob.getSpec().getDeploymentName()),
validateJobNotEmpty(sessionJob),
validateNotLastStateUpgradeMode(sessionJob),
validateSpecChange(sessionJob),
@@ -384,11 +378,7 @@ public class DefaultValidator implements
FlinkResourceValidator {
return firstPresent(
validateNotApplicationCluster(sessionCluster),
validateSessionClusterId(sessionJob, sessionCluster),
- validateJobSpec(
- sessionJob.getSpec().getJob(),
- null,
- effectiveConfig,
-
KubernetesDeploymentMode.getDeploymentMode(sessionCluster)));
+ validateJobSpec(sessionJob.getSpec().getJob(), null,
effectiveConfig));
}
private Optional<String> validateJobNotEmpty(FlinkSessionJob sessionJob) {
@@ -430,16 +420,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 77a6c44b..4e47e0aa 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
@@ -495,6 +495,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(
