You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gy...@apache.org on 2023/06/27 15:38:17 UTC
[flink-kubernetes-operator] branch main updated: [FLINK-32012] Provide rollback feature on savepoint upgrade mode if HA missing and JM never started
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 d346ca9c [FLINK-32012] Provide rollback feature on savepoint upgrade mode if HA missing and JM never started
d346ca9c is described below
commit d346ca9c437d20042ed8f4a1954f0f0ed438b3ae
Author: Nicolas Fraison <64...@users.noreply.github.com>
AuthorDate: Tue Jun 27 17:38:11 2023 +0200
[FLINK-32012] Provide rollback feature on savepoint upgrade mode if HA missing and JM never started
---
.../operator/reconciler/ReconciliationUtils.java | 23 +++-
.../AbstractFlinkResourceReconciler.java | 70 ++++++++----
.../deployment/AbstractJobReconciler.java | 33 +-----
.../reconciler/deployment/SessionReconciler.java | 16 ---
.../kubernetes/operator/TestingFlinkService.java | 3 +
.../operator/controller/RollbackTest.java | 125 ++++++++++++++++++++-
6 files changed, 199 insertions(+), 71 deletions(-)
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconciliationUtils.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconciliationUtils.java
index c16474a1..ab4e32e7 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconciliationUtils.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconciliationUtils.java
@@ -123,12 +123,23 @@ public class ReconciliationUtils {
// Clear errors
status.setError(null);
reconciliationStatus.setReconciliationTimestamp(clock.instant().toEpochMilli());
- reconciliationStatus.setState(
- upgrading ? ReconciliationState.UPGRADING : ReconciliationState.DEPLOYED);
+ ReconciliationState state;
+ if (status.getReconciliationStatus().getState() == ReconciliationState.ROLLING_BACK) {
+ state = upgrading ? ReconciliationState.ROLLING_BACK : ReconciliationState.ROLLED_BACK;
+ } else {
+ state = upgrading ? ReconciliationState.UPGRADING : ReconciliationState.DEPLOYED;
+ }
+ reconciliationStatus.setState(state);
+ SPEC clonedSpec;
+ if (status.getReconciliationStatus().getState() == ReconciliationState.ROLLING_BACK
+ || status.getReconciliationStatus().getState() == ReconciliationState.ROLLED_BACK) {
+ clonedSpec = reconciliationStatus.deserializeLastReconciledSpec();
+ } else {
+ clonedSpec = ReconciliationUtils.clone(spec);
+ }
if (spec.getJob() != null) {
// For jobs we have to adjust the reconciled spec
- var clonedSpec = ReconciliationUtils.clone(spec);
var job = clonedSpec.getJob();
job.setState(stateAfterReconcile);
@@ -151,7 +162,7 @@ public class ReconciliationUtils {
reconciliationStatus.markReconciledSpecAsStable();
}
} else {
- reconciliationStatus.serializeAndSetLastReconciledSpec(spec, target);
+ reconciliationStatus.serializeAndSetLastReconciledSpec(clonedSpec, target);
}
}
@@ -352,7 +363,9 @@ public class ReconciliationUtils {
} else {
// We need to observe/reconcile using the last version of the deployment spec
deployment.setSpec(lastReconciledSpecWithMeta.getSpec());
- if (status.getReconciliationStatus().getState() == ReconciliationState.UPGRADING) {
+ if (status.getReconciliationStatus().getState() == ReconciliationState.UPGRADING
+ || status.getReconciliationStatus().getState()
+ == ReconciliationState.ROLLING_BACK) {
// We were in the middle of an application upgrade, must set desired state to
// running
deployment.getSpec().getJob().setState(JobState.RUNNING);
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 2e3a8fb5..7635ade1 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
@@ -39,8 +39,8 @@ import org.apache.flink.kubernetes.operator.reconciler.Reconciler;
import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
import org.apache.flink.kubernetes.operator.reconciler.diff.DiffResult;
import org.apache.flink.kubernetes.operator.reconciler.diff.ReflectiveDiffBuilder;
-import org.apache.flink.kubernetes.operator.service.FlinkService;
import org.apache.flink.kubernetes.operator.utils.EventRecorder;
+import org.apache.flink.kubernetes.operator.utils.FlinkUtils;
import org.apache.flink.kubernetes.operator.utils.StatusRecorder;
import org.apache.flink.runtime.jobmanager.HighAvailabilityMode;
@@ -94,11 +94,33 @@ public abstract class AbstractFlinkResourceReconciler<
this.resourceScaler = autoscalerFactory.create(kubernetesClient, eventRecorder);
}
+ private boolean prepareCrForRollback(
+ FlinkResourceContext<CR> ctx, SPEC currentDeploySpec, SPEC lastReconciledSpec) {
+ var cr = ctx.getResource();
+ var reconciliationStatus = cr.getStatus().getReconciliationStatus();
+ // Spec has changed while rolling back we should apply new spec and move to upgrading
+ // state
+ // Don't take in account changes on job.state as it could be overriden to running if the
+ // current spec is not valid
+ if (lastReconciledSpec.getJob() != null) {
+ lastReconciledSpec.getJob().setState(currentDeploySpec.getJob().getState());
+ }
+ var specDiffRollingBack =
+ new ReflectiveDiffBuilder<>(
+ ctx.getDeploymentMode(), lastReconciledSpec, currentDeploySpec)
+ .build();
+ if (DiffType.IGNORE != specDiffRollingBack.getType()) {
+ reconciliationStatus.setState(ReconciliationState.UPGRADING);
+ } else {
+ // Rely on the last stable spec if rolling back and no change in the spec
+ cr.setSpec(cr.getStatus().getReconciliationStatus().deserializeLastStableSpec());
+ }
+ return true;
+ }
+
@Override
public void reconcile(FlinkResourceContext<CR> ctx) throws Exception {
var cr = ctx.getResource();
- var spec = cr.getSpec();
- var deployConfig = ctx.getDeployConfig(spec);
var status = cr.getStatus();
var reconciliationStatus = cr.getStatus().getReconciliationStatus();
@@ -112,6 +134,8 @@ public abstract class AbstractFlinkResourceReconciler<
// No further logic is required at this point.
if (reconciliationStatus.isBeforeFirstDeployment()) {
LOG.info("Deploying for the first time");
+ var spec = cr.getSpec();
+ var deployConfig = ctx.getDeployConfig(spec);
updateStatusBeforeFirstDeployment(cr, spec, deployConfig, status);
deploy(
ctx,
@@ -138,14 +162,17 @@ public abstract class AbstractFlinkResourceReconciler<
DiffType.IGNORE != diffType
|| reconciliationStatus.getState() == ReconciliationState.UPGRADING;
+ if (reconciliationStatus.getState() == ReconciliationState.ROLLING_BACK) {
+ specChanged = prepareCrForRollback(ctx, currentDeploySpec, lastReconciledSpec);
+ }
+
var observeConfig = ctx.getObserveConfig();
if (specChanged) {
+ var deployConfig = ctx.getDeployConfig(cr.getSpec());
if (checkNewSpecAlreadyDeployed(cr, deployConfig)) {
return;
}
-
triggerSpecChangeEvent(cr, specDiff);
-
boolean reconciled =
scale(ctx, deployConfig, diffType) || reconcileSpecChange(ctx, deployConfig);
if (reconciled) {
@@ -157,7 +184,7 @@ public abstract class AbstractFlinkResourceReconciler<
ReconciliationUtils.updateReconciliationMetadata(cr);
}
- if (shouldRollBack(cr, observeConfig, ctx.getFlinkService())) {
+ if (shouldRollBack(ctx, observeConfig)) {
// Rollbacks are executed in two steps, we initiate it first then return
if (initiateRollBack(status)) {
return;
@@ -169,7 +196,6 @@ public abstract class AbstractFlinkResourceReconciler<
EventRecorder.Reason.Rollback,
EventRecorder.Component.JobManagerDeployment,
MSG_ROLLBACK);
- rollback(ctx);
} else if (!reconcileOtherChanges(ctx)) {
if (!resourceScaler.scale(ctx)) {
LOG.info("Resource fully reconciled, nothing to do...");
@@ -238,14 +264,6 @@ public abstract class AbstractFlinkResourceReconciler<
protected abstract boolean reconcileSpecChange(
FlinkResourceContext<CR> ctx, Configuration deployConfig) throws Exception;
- /**
- * Rollback deployed resource to the last stable spec.
- *
- * @param ctx Reconciliation context.
- * @throws Exception Error during rollback.
- */
- protected abstract void rollback(FlinkResourceContext<CR> ctx) throws Exception;
-
/**
* Reconcile any other changes required for this resource that are specific to the reconciler
* implementation.
@@ -299,7 +317,9 @@ public abstract class AbstractFlinkResourceReconciler<
*/
private boolean checkNewSpecAlreadyDeployed(CR resource, Configuration deployConf) {
if (resource.getStatus().getReconciliationStatus().getState()
- == ReconciliationState.UPGRADING) {
+ == ReconciliationState.UPGRADING
+ || resource.getStatus().getReconciliationStatus().getState()
+ == ReconciliationState.ROLLING_BACK) {
return false;
}
AbstractFlinkSpec deployedSpec = ReconciliationUtils.getDeployedSpec(resource);
@@ -344,15 +364,13 @@ public abstract class AbstractFlinkResourceReconciler<
*
* <p>Rollbacks are only supported to previously running resource specs with HA enabled.
*
- * @param resource Resource being reconciled.
+ * @param ctx Reconciliation context.
* @param configuration Flink cluster configuration.
* @return True if the resource should be rolled back.
*/
- private boolean shouldRollBack(
- AbstractFlinkResource<SPEC, STATUS> resource,
- Configuration configuration,
- FlinkService flinkService) {
+ private boolean shouldRollBack(FlinkResourceContext<CR> ctx, Configuration configuration) {
+ var resource = ctx.getResource();
var reconciliationStatus = resource.getStatus().getReconciliationStatus();
if (reconciliationStatus.getState() == ReconciliationState.ROLLING_BACK) {
return true;
@@ -389,7 +407,15 @@ public abstract class AbstractFlinkResourceReconciler<
return false;
}
- var haDataAvailable = flinkService.isHaMetadataAvailable(configuration);
+ if (resource.getSpec().getJob() != null
+ && resource.getSpec().getJob().getUpgradeMode() == UpgradeMode.SAVEPOINT
+ && FlinkUtils.jmPodNeverStarted(ctx.getJosdkContext())) {
+ // HA data not available as JM never start and relying on SAVEPOINT upgrade mode
+ // Safe to rollback relying on savepoint
+ return true;
+ }
+
+ var haDataAvailable = ctx.getFlinkService().isHaMetadataAvailable(configuration);
if (!haDataAvailable) {
LOG.warn("Rollback is not possible due to missing HA metadata");
}
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 b20f0074..dce52aef 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
@@ -26,7 +26,6 @@ import org.apache.flink.kubernetes.operator.api.spec.JobState;
import org.apache.flink.kubernetes.operator.api.spec.UpgradeMode;
import org.apache.flink.kubernetes.operator.api.status.CommonStatus;
import org.apache.flink.kubernetes.operator.api.status.JobStatus;
-import org.apache.flink.kubernetes.operator.api.status.ReconciliationState;
import org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
import org.apache.flink.kubernetes.operator.controller.FlinkResourceContext;
import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
@@ -113,9 +112,13 @@ public abstract class AbstractJobReconciler<
EventRecorder.Reason.Suspended,
EventRecorder.Component.JobManagerDeployment,
MSG_SUSPENDED);
+
+ UpgradeMode upgradeMode = availableUpgradeMode.getUpgradeMode().get();
+
// We must record the upgrade mode used to the status later
- currentDeploySpec.getJob().setUpgradeMode(availableUpgradeMode.getUpgradeMode().get());
- cancelJob(ctx, availableUpgradeMode.getUpgradeMode().get());
+ currentDeploySpec.getJob().setUpgradeMode(upgradeMode);
+
+ cancelJob(ctx, upgradeMode);
if (desiredJobState == JobState.RUNNING) {
ReconciliationUtils.updateStatusBeforeDeploymentAttempt(
resource, deployConfig, clock);
@@ -261,30 +264,6 @@ public abstract class AbstractJobReconciler<
deploy(ctx, spec, deployConfig, savepointOpt, requireHaMetadata);
}
- @Override
- protected void rollback(FlinkResourceContext<CR> ctx) throws Exception {
- var resource = ctx.getResource();
- var reconciliationStatus = resource.getStatus().getReconciliationStatus();
- var rollbackSpec = reconciliationStatus.deserializeLastStableSpec();
- rollbackSpec.getJob().setUpgradeMode(UpgradeMode.LAST_STATE);
-
- UpgradeMode upgradeMode = resource.getSpec().getJob().getUpgradeMode();
-
- cancelJob(
- ctx,
- upgradeMode == UpgradeMode.STATELESS
- ? UpgradeMode.STATELESS
- : UpgradeMode.LAST_STATE);
-
- restoreJob(
- ctx,
- rollbackSpec,
- ctx.getDeployConfig(rollbackSpec),
- upgradeMode != UpgradeMode.STATELESS);
-
- reconciliationStatus.setState(ReconciliationState.ROLLED_BACK);
- }
-
@Override
public boolean reconcileOtherChanges(FlinkResourceContext<CR> ctx) throws Exception {
var status = ctx.getResource().getStatus();
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconciler.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconciler.java
index eb486a24..4e3ede17 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconciler.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/SessionReconciler.java
@@ -23,8 +23,6 @@ import org.apache.flink.kubernetes.operator.api.FlinkSessionJob;
import org.apache.flink.kubernetes.operator.api.spec.FlinkDeploymentSpec;
import org.apache.flink.kubernetes.operator.api.status.FlinkDeploymentStatus;
import org.apache.flink.kubernetes.operator.api.status.JobManagerDeploymentStatus;
-import org.apache.flink.kubernetes.operator.api.status.ReconciliationState;
-import org.apache.flink.kubernetes.operator.api.status.ReconciliationStatus;
import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
import org.apache.flink.kubernetes.operator.controller.FlinkResourceContext;
import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
@@ -106,20 +104,6 @@ public class SessionReconciler
IngressUtils.updateIngressRules(cr.getMetadata(), spec, deployConfig, kubernetesClient);
}
- @Override
- protected void rollback(FlinkResourceContext<FlinkDeployment> ctx) throws Exception {
- var deployment = ctx.getResource();
- FlinkDeploymentStatus status = deployment.getStatus();
- ReconciliationStatus<FlinkDeploymentSpec> reconciliationStatus =
- status.getReconciliationStatus();
- FlinkDeploymentSpec rollbackSpec = reconciliationStatus.deserializeLastStableSpec();
-
- deleteSessionCluster(ctx);
- deploy(ctx, rollbackSpec, ctx.getDeployConfig(rollbackSpec), Optional.empty(), false);
-
- reconciliationStatus.setState(ReconciliationState.ROLLED_BACK);
- }
-
@Override
public boolean reconcileOtherChanges(FlinkResourceContext<FlinkDeployment> ctx)
throws Exception {
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java
index 22d6be41..4309da43 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java
@@ -125,6 +125,8 @@ public class TestingFlinkService extends AbstractFlinkService {
@Getter private int desiredReplicas = 0;
@Getter private int cancelJobCallCount = 0;
+ @Getter private Configuration submittedConf;
+
@Setter
private Tuple2<
Optional<CheckpointHistoryWrapper.CompletedCheckpointInfo>,
@@ -178,6 +180,7 @@ public class TestingFlinkService extends AbstractFlinkService {
validateHaMetadataExists(conf);
}
deployApplicationCluster(jobSpec, removeOperatorConfigs(conf));
+ submittedConf = conf.clone();
}
protected void deployApplicationCluster(JobSpec jobSpec, Configuration conf) throws Exception {
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/RollbackTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/RollbackTest.java
index ecce2487..8d63b63d 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/RollbackTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/RollbackTest.java
@@ -89,6 +89,7 @@ public class RollbackTest {
dep,
() -> {
dep.getSpec().getJob().setParallelism(9999);
+ dep.getSpec().getFlinkConfiguration().put("test.deploy.config", "roll_back");
testController.reconcile(dep, context);
assertEquals(
JobState.SUSPENDED,
@@ -98,6 +99,11 @@ public class RollbackTest {
.getJob()
.getState());
testController.reconcile(dep, context);
+ assertEquals(
+ "roll_back",
+ flinkService
+ .getSubmittedConf()
+ .getString("test.deploy.config", "unknown"));
// Trigger rollback by delaying the recovery
offsetReconcilerClock(dep, Duration.ofSeconds(15));
@@ -112,6 +118,98 @@ public class RollbackTest {
true);
}
+ @Test
+ public void testSavepointRollbackWithoutHaMetadata() throws Exception {
+ flinkService.setHaDataAvailable(false);
+ var dep = TestUtils.buildApplicationCluster();
+ dep.getSpec().getJob().setUpgradeMode(UpgradeMode.SAVEPOINT);
+ offsetReconcilerClock(dep, Duration.ZERO);
+
+ testRollback(
+ dep,
+ () -> {
+ dep.getSpec().getJob().setParallelism(9999);
+ dep.getSpec().getFlinkConfiguration().put("test.deploy.config", "roll_back");
+ testController.reconcile(dep, context);
+ assertEquals(
+ JobState.SUSPENDED,
+ dep.getStatus()
+ .getReconciliationStatus()
+ .deserializeLastReconciledSpec()
+ .getJob()
+ .getState());
+ testController.reconcile(dep, context);
+ assertEquals(
+ "roll_back",
+ flinkService
+ .getSubmittedConf()
+ .getString("test.deploy.config", "unknown"));
+
+ // Trigger rollback by delaying the recovery
+ offsetReconcilerClock(dep, Duration.ofSeconds(15));
+
+ // Update JM deployment status to simulate JM never start
+ flinkService.setJobManagerReady(false);
+
+ testController.reconcile(dep, context);
+ },
+ () -> {
+ assertEquals("RUNNING", dep.getStatus().getJobStatus().getState());
+ assertEquals(1, flinkService.listJobs().size());
+ dep.getSpec().setRestartNonce(10L);
+ testController.reconcile(dep, context);
+ },
+ true);
+ }
+
+ @Test
+ public void testSavepointNoRollbackWithoutHaMetadataAndJMWasReady() throws Exception {
+ flinkService.setHaDataAvailable(false);
+ var deployment = TestUtils.buildApplicationCluster();
+ deployment.getSpec().getJob().setUpgradeMode(UpgradeMode.SAVEPOINT);
+ offsetReconcilerClock(deployment, Duration.ZERO);
+
+ var flinkConfiguration = deployment.getSpec().getFlinkConfiguration();
+ flinkConfiguration.put(
+ KubernetesOperatorConfigOptions.DEPLOYMENT_ROLLBACK_ENABLED.key(), "true");
+ flinkConfiguration.put(
+ KubernetesOperatorConfigOptions.DEPLOYMENT_READINESS_TIMEOUT.key(), "10s");
+
+ testController.reconcile(deployment, context);
+
+ // Validate reconciliation status
+ testController.reconcile(deployment, context);
+ testController.reconcile(deployment, context);
+
+ // Validate stable job
+ assertTrue(deployment.getStatus().getReconciliationStatus().isLastReconciledSpecStable());
+
+ deployment.getSpec().getJob().setParallelism(9999);
+ testController.reconcile(deployment, context);
+ assertEquals(
+ JobState.SUSPENDED,
+ deployment
+ .getStatus()
+ .getReconciliationStatus()
+ .deserializeLastReconciledSpec()
+ .getJob()
+ .getState());
+ testController.reconcile(deployment, context);
+
+ // Trigger rollback by delaying the recovery
+ offsetReconcilerClock(deployment, Duration.ofSeconds(15));
+
+ testController.reconcile(deployment, context);
+
+ assertFalse(deployment.getStatus().getReconciliationStatus().isLastReconciledSpecStable());
+ assertEquals(
+ ReconciliationState.DEPLOYED,
+ deployment.getStatus().getReconciliationStatus().getState());
+ assertEquals(
+ JobManagerDeploymentStatus.DEPLOYED_NOT_READY,
+ deployment.getStatus().getJobManagerDeploymentStatus());
+ }
+
@Test
public void testRollbackFailureWithLastState() throws Exception {
var dep = TestUtils.buildApplicationCluster();
@@ -123,6 +221,7 @@ public class RollbackTest {
dep,
() -> {
dep.getSpec().getJob().setParallelism(9999);
+ dep.getSpec().getFlinkConfiguration().put("test.deploy.config", "roll_back");
dep.getSpec().getFlinkConfiguration().remove("t");
testController.reconcile(dep, context);
assertEquals(
@@ -133,6 +232,11 @@ public class RollbackTest {
.getJob()
.getState());
testController.reconcile(dep, context);
+ assertEquals(
+ "roll_back",
+ flinkService
+ .getSubmittedConf()
+ .getString("test.deploy.config", "unknown"));
// Trigger rollback by delaying the recovery
offsetReconcilerClock(dep, Duration.ofSeconds(15));
@@ -180,6 +284,7 @@ public class RollbackTest {
.key(),
"false");
dep.getSpec().getJob().setParallelism(9999);
+ dep.getSpec().getFlinkConfiguration().put("test.deploy.config", "roll_back");
testController.reconcile(dep, context);
assertEquals(
JobState.SUSPENDED,
@@ -189,6 +294,11 @@ public class RollbackTest {
.getJob()
.getState());
testController.reconcile(dep, context);
+ assertEquals(
+ "roll_back",
+ flinkService
+ .getSubmittedConf()
+ .getString("test.deploy.config", "unknown"));
// Validate that rollback config is picked up from latest deploy conf
dep.getSpec()
.getFlinkConfiguration()
@@ -251,6 +361,7 @@ public class RollbackTest {
KubernetesOperatorConfigOptions.DEPLOYMENT_ROLLBACK_ENABLED.key(), "true");
flinkConfiguration.put(
KubernetesOperatorConfigOptions.DEPLOYMENT_READINESS_TIMEOUT.key(), "10s");
+ flinkConfiguration.put("test.deploy.config", "stable");
testController.reconcile(deployment, context);
@@ -275,11 +386,18 @@ public class RollbackTest {
if (injectValidationError) {
deployment.getSpec().setLogConfiguration(Map.of("invalid", "entry"));
}
-
+ flinkService.setJobManagerReady(true);
+ testController.reconcile(deployment, context);
testController.reconcile(deployment, context);
assertEquals(
ReconciliationState.ROLLED_BACK,
deployment.getStatus().getReconciliationStatus().getState());
+ if (flinkService.getSubmittedConf() != null) {
+ assertEquals(
+ "stable",
+ flinkService.getSubmittedConf().getString("test.deploy.config", "unknown"));
+ }
+
deployment.getSpec().setLogConfiguration(null);
testController.reconcile(deployment, context);
@@ -306,6 +424,9 @@ public class RollbackTest {
deployment.getSpec().setRestartNonce(456L);
triggerRollback.run();
+
+ testController.reconcile(deployment, context);
+ flinkService.setJobManagerReady(true);
testController.reconcile(deployment, context);
assertEquals(
ReconciliationState.ROLLED_BACK,
@@ -317,6 +438,7 @@ public class RollbackTest {
deployment.setSpec(
deployment.getStatus().getReconciliationStatus().deserializeLastStableSpec());
testController.reconcile(deployment, context);
+ testController.reconcile(deployment, context);
assertEquals(
ReconciliationState.DEPLOYED,
deployment.getStatus().getReconciliationStatus().getState());
@@ -352,6 +474,7 @@ public class RollbackTest {
// Verify suspending a rolled back job
triggerRollback.run();
testController.reconcile(deployment, context);
+ testController.reconcile(deployment, context);
assertEquals(
ReconciliationState.ROLLED_BACK,
deployment.getStatus().getReconciliationStatus().getState());