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 1eca2c5e [FLINK-28755] Fix error when switching from stateless to
savepoint upgrade mode
1eca2c5e is described below
commit 1eca2c5e6a62bd9f1c9e752191f8a7477903d73c
Author: Gabor Somogyi <[email protected]>
AuthorDate: Wed Sep 14 09:43:42 2022 +0200
[FLINK-28755] Fix error when switching from stateless to savepoint upgrade
mode
---
.../operator/config/FlinkConfigManager.java | 16 +-
.../deployment/AbstractJobReconciler.java | 6 +-
.../deployment/ApplicationReconcilerTest.java | 150 -------
.../ApplicationReconcilerUpgradeModeTest.java | 457 +++++++++++++++++++++
4 files changed, 475 insertions(+), 154 deletions(-)
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManager.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManager.java
index 49181886..5f24df7c 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManager.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManager.java
@@ -19,6 +19,7 @@
package org.apache.flink.kubernetes.operator.config;
import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.GlobalConfiguration;
import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
@@ -53,6 +54,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;
+import static
org.apache.flink.configuration.CheckpointingOptions.SAVEPOINT_DIRECTORY;
import static
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.K8S_OP_CONF_PREFIX;
import static
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.OPERATOR_DYNAMIC_CONFIG_CHECK_INTERVAL;
import static
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.OPERATOR_DYNAMIC_CONFIG_ENABLED;
@@ -160,7 +162,7 @@ public class FlinkConfigManager {
"Cannot create observe config before first deployment,
this indicates a bug.");
}
var conf = getConfig(deployment.getMetadata(), deployedSpec);
- addOperatorConfigsFromSpec(deployment.getSpec(), conf);
+ applyConfigsFromCurrentSpec(deployment.getSpec(), conf,
SAVEPOINT_DIRECTORY);
return conf;
}
@@ -177,6 +179,18 @@ public class FlinkConfigManager {
}
}
+ @SuppressWarnings("unchecked")
+ private void applyConfigsFromCurrentSpec(
+ AbstractFlinkSpec spec, Configuration conf, ConfigOption...
configOptions) {
+ addOperatorConfigsFromSpec(spec, conf);
+ if (spec.getFlinkConfiguration() != null) {
+ var deployConfig =
Configuration.fromMap(spec.getFlinkConfiguration());
+ for (ConfigOption configOption : configOptions) {
+ deployConfig.getOptional(configOption).ifPresent(v ->
conf.set(configOption, v));
+ }
+ }
+ }
+
public Configuration getSessionJobConfig(
FlinkDeployment deployment, FlinkSessionJobSpec sessionJobSpec) {
Configuration sessionJobConfig = getObserveConfig(deployment);
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractJobReconciler.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractJobReconciler.java
index 4f850abe..11a9ac74 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractJobReconciler.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractJobReconciler.java
@@ -154,9 +154,6 @@ public abstract class AbstractJobReconciler<
CR resource, Configuration deployConfig, Configuration
observeConfig) {
var status = resource.getStatus();
var upgradeMode = resource.getSpec().getJob().getUpgradeMode();
- var changedToLastStateWithoutHa =
-
ReconciliationUtils.isUpgradeModeChangedToLastStateAndHADisabledPreviously(
- resource, observeConfig);
if (upgradeMode == UpgradeMode.STATELESS) {
LOG.info("Stateless job, ready for upgrade");
@@ -171,6 +168,9 @@ public abstract class AbstractJobReconciler<
if (ReconciliationUtils.isJobRunning(status)) {
LOG.info("Job is in running state, ready for upgrade with {}",
upgradeMode);
+ var changedToLastStateWithoutHa =
+
ReconciliationUtils.isUpgradeModeChangedToLastStateAndHADisabledPreviously(
+ resource, observeConfig);
if (changedToLastStateWithoutHa) {
LOG.info(
"Using savepoint upgrade mode when switching to
last-state without HA previously enabled");
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 b8f5e9af..b102cdb1 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
@@ -189,78 +189,6 @@ public class ApplicationReconcilerTest {
assertEquals("finished_sp", runningJobs.get(0).f0);
}
- @ParameterizedTest
- @EnumSource(UpgradeMode.class)
- public void testUpgradeBeforeReachingStableSpec(UpgradeMode upgradeMode)
throws Exception {
- flinkService.setHaDataAvailable(false);
-
- final FlinkDeployment deployment = TestUtils.buildApplicationCluster();
-
- reconciler.reconcile(deployment, context);
- assertEquals(
- JobManagerDeploymentStatus.DEPLOYING,
- deployment.getStatus().getJobManagerDeploymentStatus());
-
- // Ready for spec changes, the reconciliation should be performed
- final String newImage = "new-image-1";
- deployment.getSpec().getJob().setUpgradeMode(upgradeMode);
- deployment.getSpec().setImage(newImage);
- reconciler.reconcile(deployment, context);
- if (!UpgradeMode.STATELESS.equals(upgradeMode)) {
-
assertNull(deployment.getStatus().getReconciliationStatus().getLastReconciledSpec());
- assertEquals(
- ReconciliationState.UPGRADING,
-
deployment.getStatus().getReconciliationStatus().getState());
- reconciler.reconcile(deployment, context);
- }
- assertEquals(
- newImage,
- deployment
- .getStatus()
- .getReconciliationStatus()
- .deserializeLastReconciledSpec()
- .getImage());
- }
-
- @Test
- public void testUpgradeModeChangeFromSavepointToLastState() throws
Exception {
- final String expectedSavepointPath = "savepoint_0";
- final FlinkDeployment deployment = TestUtils.buildApplicationCluster();
-
- reconciler.reconcile(deployment, context);
- var runningJobs = flinkService.listJobs();
- verifyAndSetRunningJobsToStatus(deployment, runningJobs);
-
- // Suspend FlinkDeployment with savepoint upgrade mode
- deployment.getSpec().getJob().setUpgradeMode(UpgradeMode.SAVEPOINT);
- deployment.getSpec().getJob().setState(JobState.SUSPENDED);
- deployment.getSpec().setImage("new-image-1");
-
- reconciler.reconcile(deployment, context);
- assertEquals(0, flinkService.getRunningCount());
- assertEquals(
- org.apache.flink.api.common.JobStatus.FINISHED.name(),
- deployment.getStatus().getJobStatus().getState());
-
- assertEquals(
- expectedSavepointPath,
- deployment
- .getStatus()
- .getJobStatus()
- .getSavepointInfo()
- .getLastSavepoint()
- .getLocation());
-
- deployment.getSpec().getJob().setUpgradeMode(UpgradeMode.LAST_STATE);
- deployment.getSpec().getJob().setState(JobState.RUNNING);
- deployment.getSpec().setImage("new-image-2");
-
- reconciler.reconcile(deployment, context);
- runningJobs = flinkService.listJobs();
- assertEquals(1, flinkService.getRunningCount());
- assertEquals(expectedSavepointPath, runningJobs.get(0).f0);
- }
-
@Test
public void triggerSavepoint() throws Exception {
FlinkDeployment deployment = TestUtils.buildApplicationCluster();
@@ -358,84 +286,6 @@ public class ApplicationReconcilerTest {
assertTrue(SavepointUtils.savepointInProgress(spDeployment.getStatus().getJobStatus()));
}
- @Test
- public void
testUpgradeModeChangedToLastStateShouldTriggerSavepointWhileHADisabled()
- throws Exception {
- flinkService.setHaDataAvailable(false);
-
- final FlinkDeployment deployment = TestUtils.buildApplicationCluster();
-
deployment.getSpec().getFlinkConfiguration().remove(HighAvailabilityOptions.HA_MODE.key());
-
- reconciler.reconcile(deployment, context);
- assertEquals(
- JobManagerDeploymentStatus.DEPLOYING,
- deployment.getStatus().getJobManagerDeploymentStatus());
-
- // Not ready for spec changes, the reconciliation is not performed
- final String newImage = "new-image-1";
- deployment.getSpec().getJob().setUpgradeMode(UpgradeMode.LAST_STATE);
- deployment.getSpec().setImage(newImage);
- reconciler.reconcile(deployment, context);
- reconciler.reconcile(deployment, context);
- assertNull(flinkService.listJobs().get(0).f0);
- assertNotEquals(
- newImage,
- deployment
- .getStatus()
- .getReconciliationStatus()
- .deserializeLastReconciledSpec()
- .getImage());
-
- // Ready for spec changes, the reconciliation should be performed
- verifyAndSetRunningJobsToStatus(deployment, flinkService.listJobs());
- reconciler.reconcile(deployment, context);
- reconciler.reconcile(deployment, context);
- assertEquals(
- newImage,
- deployment
- .getStatus()
- .getReconciliationStatus()
- .deserializeLastReconciledSpec()
- .getImage());
- // Upgrade mode changes from stateless to last-state should trigger a
savepoint
- final String expectedSavepointPath = "savepoint_0";
- var runningJobs = flinkService.listJobs();
- assertEquals(expectedSavepointPath, runningJobs.get(0).f0);
- }
-
- @Test
- public void
testUpgradeModeChangedToLastStateShouldNotTriggerSavepointWhileHAEnabled()
- throws Exception {
- final FlinkDeployment deployment = TestUtils.buildApplicationCluster();
-
- reconciler.reconcile(deployment, context);
- assertNotEquals(
- UpgradeMode.LAST_STATE,
- deployment
- .getStatus()
- .getReconciliationStatus()
- .deserializeLastReconciledSpec()
- .getJob()
- .getUpgradeMode());
-
- final String newImage = "new-image-1";
- deployment.getSpec().getJob().setUpgradeMode(UpgradeMode.LAST_STATE);
- deployment.getSpec().setImage(newImage);
- verifyAndSetRunningJobsToStatus(deployment, flinkService.listJobs());
- reconciler.reconcile(deployment, context);
- reconciler.reconcile(deployment, context);
- assertEquals(
- newImage,
- deployment
- .getStatus()
- .getReconciliationStatus()
- .deserializeLastReconciledSpec()
- .getImage());
- // Upgrade mode changes from stateless to last-state while HA enabled
previously should not
- // trigger a savepoint
- assertNull(flinkService.listJobs().get(0).f0);
- }
-
@Test
public void triggerRestart() throws Exception {
FlinkDeployment deployment = TestUtils.buildApplicationCluster();
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerUpgradeModeTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerUpgradeModeTest.java
new file mode 100644
index 00000000..5f79fc3e
--- /dev/null
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerUpgradeModeTest.java
@@ -0,0 +1,457 @@
+/*
+ * 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.reconciler.deployment;
+
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.configuration.CheckpointingOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.HighAvailabilityOptions;
+import org.apache.flink.kubernetes.operator.TestUtils;
+import org.apache.flink.kubernetes.operator.TestingFlinkService;
+import org.apache.flink.kubernetes.operator.TestingStatusRecorder;
+import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
+import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
+import org.apache.flink.kubernetes.operator.crd.spec.FlinkVersion;
+import org.apache.flink.kubernetes.operator.crd.spec.JobState;
+import org.apache.flink.kubernetes.operator.crd.spec.UpgradeMode;
+import org.apache.flink.kubernetes.operator.crd.status.FlinkDeploymentStatus;
+import
org.apache.flink.kubernetes.operator.crd.status.JobManagerDeploymentStatus;
+import org.apache.flink.kubernetes.operator.crd.status.JobStatus;
+import org.apache.flink.kubernetes.operator.crd.status.ReconciliationState;
+import org.apache.flink.kubernetes.operator.crd.status.Savepoint;
+import org.apache.flink.kubernetes.operator.crd.status.SavepointTriggerType;
+import
org.apache.flink.kubernetes.operator.exception.DeploymentFailedException;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.utils.EventRecorder;
+import org.apache.flink.runtime.client.JobStatusMessage;
+
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.fail;
+
+/** @link JobStatusObserver unit tests */
+@EnableKubernetesMockClient(crud = true)
+public class ApplicationReconcilerUpgradeModeTest {
+
+ private KubernetesClient kubernetesClient;
+ private final FlinkConfigManager configManager = new
FlinkConfigManager(new Configuration());
+ private TestingFlinkService flinkService;
+ private ApplicationReconciler reconciler;
+ private Context<FlinkDeployment> context;
+
+ @BeforeEach
+ public void before() {
+
kubernetesClient.resource(TestUtils.buildApplicationCluster()).createOrReplace();
+ var eventRecorder = new EventRecorder(kubernetesClient, (r, e) -> {});
+ var statusRecoder = new TestingStatusRecorder<FlinkDeployment,
FlinkDeploymentStatus>();
+ flinkService = new TestingFlinkService(kubernetesClient);
+ context = flinkService.getContext();
+ reconciler =
+ new ApplicationReconciler(
+ kubernetesClient,
+ flinkService,
+ configManager,
+ eventRecorder,
+ statusRecoder);
+ }
+
+ @ParameterizedTest
+ @EnumSource(FlinkVersion.class)
+ public void testUpgradeFromStatelessToStateless(FlinkVersion flinkVersion)
throws Exception {
+ testUpgradeToStateless(flinkVersion, UpgradeMode.STATELESS);
+ }
+
+ @ParameterizedTest
+ @EnumSource(FlinkVersion.class)
+ public void testUpgradeFromSavepointToStateless(FlinkVersion flinkVersion)
throws Exception {
+ testUpgradeToStateless(flinkVersion, UpgradeMode.SAVEPOINT);
+ }
+
+ @ParameterizedTest
+ @EnumSource(FlinkVersion.class)
+ public void testUpgradeFromLastStateToStateless(FlinkVersion flinkVersion)
throws Exception {
+ testUpgradeToStateless(flinkVersion, UpgradeMode.LAST_STATE);
+ }
+
+ private void testUpgradeToStateless(FlinkVersion flinkVersion, UpgradeMode
fromUpgradeMode)
+ throws Exception {
+ FlinkDeployment deployment = buildApplicationCluster(flinkVersion,
fromUpgradeMode);
+
+ reconciler.reconcile(deployment, context);
+ var runningJobs = flinkService.listJobs();
+ verifyAndSetRunningJobsToStatus(deployment, runningJobs);
+
+ FlinkDeployment modifiedDeployment =
+ cloneDeploymentWithUpgradeMode(deployment,
UpgradeMode.STATELESS);
+
+ reconciler.reconcile(modifiedDeployment, context);
+ assertEquals(0, flinkService.getRunningCount());
+
+ reconciler.reconcile(modifiedDeployment, context);
+ runningJobs = flinkService.listJobs();
+ assertEquals(1, flinkService.getRunningCount());
+ assertNull(runningJobs.get(0).f0);
+ }
+
+ @ParameterizedTest
+ @EnumSource(FlinkVersion.class)
+ public void testUpgradeFromStatelessToSavepoint(FlinkVersion flinkVersion)
throws Exception {
+ testUpgradeToSavepoint(flinkVersion, UpgradeMode.STATELESS);
+ }
+
+ @ParameterizedTest
+ @EnumSource(FlinkVersion.class)
+ public void testUpgradeFromSavepointToSavepoint(FlinkVersion flinkVersion)
throws Exception {
+ testUpgradeToSavepoint(flinkVersion, UpgradeMode.SAVEPOINT);
+ }
+
+ @ParameterizedTest
+ @EnumSource(FlinkVersion.class)
+ public void testUpgradeFromLastStateToSavepoint(FlinkVersion flinkVersion)
throws Exception {
+ testUpgradeToSavepoint(flinkVersion, UpgradeMode.LAST_STATE);
+ }
+
+ private void testUpgradeToSavepoint(FlinkVersion flinkVersion, UpgradeMode
fromUpgradeMode)
+ throws Exception {
+ FlinkDeployment deployment = buildApplicationCluster(flinkVersion,
fromUpgradeMode);
+
+ reconciler.reconcile(deployment, context);
+ var runningJobs = flinkService.listJobs();
+ verifyAndSetRunningJobsToStatus(deployment, runningJobs);
+
+ FlinkDeployment modifiedDeployment =
+ cloneDeploymentWithUpgradeMode(deployment,
UpgradeMode.SAVEPOINT);
+ modifiedDeployment
+ .getSpec()
+ .getFlinkConfiguration()
+ .put(CheckpointingOptions.SAVEPOINT_DIRECTORY.key(),
"test-savepoint-dir");
+
+ reconciler.reconcile(modifiedDeployment, context);
+ assertEquals(0, flinkService.getRunningCount());
+
+ reconciler.reconcile(modifiedDeployment, context);
+ runningJobs = flinkService.listJobs();
+ assertEquals(1, flinkService.getRunningCount());
+ assertEquals("savepoint_0", runningJobs.get(0).f0);
+ assertEquals(
+ SavepointTriggerType.UPGRADE,
+ modifiedDeployment
+ .getStatus()
+ .getJobStatus()
+ .getSavepointInfo()
+ .getLastSavepoint()
+ .getTriggerType());
+ }
+
+ @ParameterizedTest
+ @EnumSource(FlinkVersion.class)
+ public void testUpgradeFromStatelessToLastState(FlinkVersion flinkVersion)
throws Exception {
+ testUpgradeToLastState(flinkVersion, UpgradeMode.STATELESS);
+ }
+
+ @ParameterizedTest
+ @EnumSource(FlinkVersion.class)
+ public void testUpgradeFromSavepointToLastState(FlinkVersion flinkVersion)
throws Exception {
+ testUpgradeToLastState(flinkVersion, UpgradeMode.SAVEPOINT);
+ }
+
+ @ParameterizedTest
+ @EnumSource(FlinkVersion.class)
+ public void testUpgradeFromLastStateToLastState(FlinkVersion flinkVersion)
throws Exception {
+ testUpgradeToLastState(flinkVersion, UpgradeMode.LAST_STATE);
+ }
+
+ private void testUpgradeToLastState(FlinkVersion flinkVersion, UpgradeMode
fromUpgradeMode)
+ throws Exception {
+ FlinkDeployment deployment = buildApplicationCluster(flinkVersion,
fromUpgradeMode);
+
+ reconciler.reconcile(deployment, context);
+ var runningJobs = flinkService.listJobs();
+ verifyAndSetRunningJobsToStatus(deployment, runningJobs);
+
+ deployment.getSpec().getJob().setUpgradeMode(UpgradeMode.LAST_STATE);
+ deployment.getSpec().setRestartNonce(100L);
+ deployment
+ .getStatus()
+ .getReconciliationStatus()
+ .setLastStableSpec(
+
deployment.getStatus().getReconciliationStatus().getLastReconciledSpec());
+ flinkService.setHaDataAvailable(false);
+ deployment.getStatus().getJobStatus().setState("RECONCILING");
+
+ Assertions.assertThrows(
+ DeploymentFailedException.class,
+ () -> {
+ deployment
+ .getStatus()
+
.setJobManagerDeploymentStatus(JobManagerDeploymentStatus.MISSING);
+ reconciler.reconcile(deployment, context);
+ fail();
+ });
+
+ Assertions.assertThrows(
+ DeploymentFailedException.class,
+ () -> {
+ deployment
+ .getStatus()
+
.setJobManagerDeploymentStatus(JobManagerDeploymentStatus.ERROR);
+ reconciler.reconcile(deployment, context);
+ });
+
+ flinkService.clear();
+ deployment.getSpec().getJob().setUpgradeMode(UpgradeMode.LAST_STATE);
+ deployment.getSpec().setRestartNonce(200L);
+ flinkService.setHaDataAvailable(false);
+ deployment
+ .getStatus()
+ .getJobStatus()
+ .getSavepointInfo()
+ .setLastSavepoint(Savepoint.of("finished_sp",
SavepointTriggerType.UPGRADE));
+ deployment.getStatus().getJobStatus().setState("FINISHED");
+
deployment.getStatus().setJobManagerDeploymentStatus(JobManagerDeploymentStatus.READY);
+ deployment
+ .getSpec()
+ .getFlinkConfiguration()
+ .put(CheckpointingOptions.SAVEPOINT_DIRECTORY.key(),
"test-savepoint-dir");
+ reconciler.reconcile(deployment, context);
+ reconciler.reconcile(deployment, context);
+
+ assertEquals(1, flinkService.getRunningCount());
+ assertEquals("finished_sp", runningJobs.get(0).f0);
+ }
+
+ private FlinkDeployment cloneDeploymentWithUpgradeMode(
+ FlinkDeployment deployment, UpgradeMode upgradeMode) {
+ FlinkDeployment result = ReconciliationUtils.clone(deployment);
+
+ result.getSpec().getJob().setUpgradeMode(upgradeMode);
+ result.getSpec().getFlinkConfiguration().put("new", "conf");
+
+ return result;
+ }
+
+ @ParameterizedTest
+ @EnumSource(UpgradeMode.class)
+ public void testUpgradeBeforeReachingStableSpec(UpgradeMode upgradeMode)
throws Exception {
+ flinkService.setHaDataAvailable(false);
+
+ final FlinkDeployment deployment = TestUtils.buildApplicationCluster();
+
+ reconciler.reconcile(deployment, context);
+ assertEquals(
+ JobManagerDeploymentStatus.DEPLOYING,
+ deployment.getStatus().getJobManagerDeploymentStatus());
+
+ // Ready for spec changes, the reconciliation should be performed
+ final String newImage = "new-image-1";
+ deployment.getSpec().getJob().setUpgradeMode(upgradeMode);
+ deployment.getSpec().setImage(newImage);
+ reconciler.reconcile(deployment, context);
+ if (!UpgradeMode.STATELESS.equals(upgradeMode)) {
+
assertNull(deployment.getStatus().getReconciliationStatus().getLastReconciledSpec());
+ assertEquals(
+ ReconciliationState.UPGRADING,
+
deployment.getStatus().getReconciliationStatus().getState());
+ reconciler.reconcile(deployment, context);
+ }
+ assertEquals(
+ newImage,
+ deployment
+ .getStatus()
+ .getReconciliationStatus()
+ .deserializeLastReconciledSpec()
+ .getImage());
+ }
+
+ @Test
+ public void testUpgradeModeChangeFromSavepointToLastState() throws
Exception {
+ final String expectedSavepointPath = "savepoint_0";
+ final FlinkDeployment deployment = TestUtils.buildApplicationCluster();
+
+ reconciler.reconcile(deployment, context);
+ var runningJobs = flinkService.listJobs();
+ verifyAndSetRunningJobsToStatus(deployment, runningJobs);
+
+ // Suspend FlinkDeployment with savepoint upgrade mode
+ deployment.getSpec().getJob().setUpgradeMode(UpgradeMode.SAVEPOINT);
+ deployment.getSpec().getJob().setState(JobState.SUSPENDED);
+ deployment.getSpec().setImage("new-image-1");
+
+ reconciler.reconcile(deployment, context);
+ assertEquals(0, flinkService.getRunningCount());
+ assertEquals(
+ org.apache.flink.api.common.JobStatus.FINISHED.name(),
+ deployment.getStatus().getJobStatus().getState());
+
+ assertEquals(
+ expectedSavepointPath,
+ deployment
+ .getStatus()
+ .getJobStatus()
+ .getSavepointInfo()
+ .getLastSavepoint()
+ .getLocation());
+
+ deployment.getSpec().getJob().setUpgradeMode(UpgradeMode.LAST_STATE);
+ deployment.getSpec().getJob().setState(JobState.RUNNING);
+ deployment.getSpec().setImage("new-image-2");
+
+ reconciler.reconcile(deployment, context);
+ runningJobs = flinkService.listJobs();
+ assertEquals(1, flinkService.getRunningCount());
+ assertEquals(expectedSavepointPath, runningJobs.get(0).f0);
+ }
+
+ @Test
+ public void
testUpgradeModeChangedToLastStateShouldTriggerSavepointWhileHADisabled()
+ throws Exception {
+ flinkService.setHaDataAvailable(false);
+
+ final FlinkDeployment deployment = TestUtils.buildApplicationCluster();
+
deployment.getSpec().getFlinkConfiguration().remove(HighAvailabilityOptions.HA_MODE.key());
+
+ reconciler.reconcile(deployment, context);
+ assertEquals(
+ JobManagerDeploymentStatus.DEPLOYING,
+ deployment.getStatus().getJobManagerDeploymentStatus());
+
+ // Not ready for spec changes, the reconciliation is not performed
+ final String newImage = "new-image-1";
+ deployment.getSpec().getJob().setUpgradeMode(UpgradeMode.LAST_STATE);
+ deployment.getSpec().setImage(newImage);
+ reconciler.reconcile(deployment, context);
+ reconciler.reconcile(deployment, context);
+ assertNull(flinkService.listJobs().get(0).f0);
+ assertNotEquals(
+ newImage,
+ deployment
+ .getStatus()
+ .getReconciliationStatus()
+ .deserializeLastReconciledSpec()
+ .getImage());
+
+ // Ready for spec changes, the reconciliation should be performed
+ verifyAndSetRunningJobsToStatus(deployment, flinkService.listJobs());
+ reconciler.reconcile(deployment, context);
+ reconciler.reconcile(deployment, context);
+ assertEquals(
+ newImage,
+ deployment
+ .getStatus()
+ .getReconciliationStatus()
+ .deserializeLastReconciledSpec()
+ .getImage());
+ // Upgrade mode changes from stateless to last-state should trigger a
savepoint
+ final String expectedSavepointPath = "savepoint_0";
+ var runningJobs = flinkService.listJobs();
+ assertEquals(expectedSavepointPath, runningJobs.get(0).f0);
+ }
+
+ @Test
+ public void
testUpgradeModeChangedToLastStateShouldNotTriggerSavepointWhileHAEnabled()
+ throws Exception {
+ final FlinkDeployment deployment = TestUtils.buildApplicationCluster();
+
+ reconciler.reconcile(deployment, context);
+ assertNotEquals(
+ UpgradeMode.LAST_STATE,
+ deployment
+ .getStatus()
+ .getReconciliationStatus()
+ .deserializeLastReconciledSpec()
+ .getJob()
+ .getUpgradeMode());
+
+ final String newImage = "new-image-1";
+ deployment.getSpec().getJob().setUpgradeMode(UpgradeMode.LAST_STATE);
+ deployment.getSpec().setImage(newImage);
+ verifyAndSetRunningJobsToStatus(deployment, flinkService.listJobs());
+ reconciler.reconcile(deployment, context);
+ reconciler.reconcile(deployment, context);
+ assertEquals(
+ newImage,
+ deployment
+ .getStatus()
+ .getReconciliationStatus()
+ .deserializeLastReconciledSpec()
+ .getImage());
+ // Upgrade mode changes from stateless to last-state while HA enabled
previously should not
+ // trigger a savepoint
+ assertNull(flinkService.listJobs().get(0).f0);
+ }
+
+ public static FlinkDeployment buildApplicationCluster(
+ FlinkVersion flinkVersion, UpgradeMode upgradeMode) {
+ FlinkDeployment deployment =
TestUtils.buildApplicationCluster(flinkVersion);
+ deployment.getSpec().getJob().setUpgradeMode(upgradeMode);
+ Map<String, String> conf =
deployment.getSpec().getFlinkConfiguration();
+
+ switch (upgradeMode) {
+ case STATELESS:
+ conf.remove(HighAvailabilityOptions.HA_MODE.key());
+ conf.remove(HighAvailabilityOptions.HA_STORAGE_PATH.key());
+ conf.remove(CheckpointingOptions.SAVEPOINT_DIRECTORY.key());
+ conf.remove(CheckpointingOptions.CHECKPOINTS_DIRECTORY.key());
+ break;
+
+ case SAVEPOINT:
+ conf.remove(HighAvailabilityOptions.HA_MODE.key());
+ conf.remove(HighAvailabilityOptions.HA_STORAGE_PATH.key());
+ break;
+
+ case LAST_STATE:
+ conf.remove(CheckpointingOptions.SAVEPOINT_DIRECTORY.key());
+ break;
+
+ default:
+ throw new RuntimeException("Unsupported upgrade mode " +
upgradeMode);
+ }
+
+ return deployment;
+ }
+
+ private void verifyAndSetRunningJobsToStatus(
+ FlinkDeployment deployment,
+ List<Tuple3<String, JobStatusMessage, Configuration>> runningJobs)
{
+ assertEquals(1, runningJobs.size());
+ assertNull(runningJobs.get(0).f0);
+ deployment
+ .getStatus()
+ .setJobStatus(
+ new JobStatus()
+ .toBuilder()
+
.jobId(runningJobs.get(0).f1.getJobId().toHexString())
+ .jobName(runningJobs.get(0).f1.getJobName())
+ .state("RUNNING")
+ .build());
+
deployment.getStatus().setJobManagerDeploymentStatus(JobManagerDeploymentStatus.READY);
+ }
+}
