This is an automated email from the ASF dual-hosted git repository.
rmetzger 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 284a8d005 [FLINK-40470] Record spec before deployment attempt in
savepoint redeploy (#1216)
284a8d005 is described below
commit 284a8d005fb6dc49c97e41580df6e83cb9636906
Author: 김현 <[email protected]>
AuthorDate: Tue Oct 6 15:58:41 2026 +0900
[FLINK-40470] Record spec before deployment attempt in savepoint redeploy
(#1216)
* [FLINK-40470] Record spec before deployment attempt in savepoint redeploy
When job.savepointRedeployNonce changes, redeployWithSavepoint cancels the
running job and redeploys it from initialSavepointPath. Unlike every other
deployment path it only recorded the target spec in the status after the
deploy call succeeded. If deploy threw (transient API server error, or an
error raised after the cluster resources were already created), the new
nonce
never reached lastReconciledSpec, every subsequent reconciliation classified
the diff as SAVEPOINT_REDEPLOY again and the operator repeated the cancel +
redeploy cycle indefinitely.
Record the target spec with updateStatusBeforeDeploymentAttempt before the
deploy call, mirroring the suspend/resume path, so that a failed attempt is
retried through the regular restore path from the recorded
upgradeSavepointPath.
The suspended -> running path used to fall back to an empty state restore
whenever the spec declared upgradeMode: stateless. This discarded the
initialSavepointPath of a savepoint redeploy requested while suspended, and
would have turned the retry above into an empty state restore for stateless
jobs. A savepoint the user explicitly requested through
initialSavepointPath,
recorded as the upgrade savepoint by the redeploy, is now restored
regardless
of the upgrade mode. Resuming a job suspended in savepoint or last-state
mode
with a stateless spec still starts from empty state, as before.
Generated-by: Claude Code (Claude Fable 5.1)
* [FLINK-40470] Mark resumed savepoint redeploy stable, use recorded path
AbstractJobReconciler's resume path completes a savepoint redeploy in two
cases: the first deployment attempt failed, or the redeploy was requested
while the job was suspended. Two fixes to that path:
- Mark the reconciled spec stable after restoring from the explicitly
requested savepoint, as redeployWithSavepoint does. Without this, a
redeploy
completed through the resume path is rolled back to the pre-redeploy spec
if
the JobManager does not become ready within the readiness timeout.
Savepoint
redeploys are documented as not supporting rollbacks.
- When deciding whether the restore targets the explicitly requested
savepoint, read initialSavepointPath from the last reconciled spec, not
from
the current spec. initialSavepointPath is ignored by the spec diff, so
editing or clearing it without a new savepointRedeployNonce is not a new
operation and must not turn the recorded redeploy into an empty-state
restore.
Also document redeploys requested while suspended under "Redeploying from a
Savepoint" instead of "Suspending and Resuming".
Generated-by: Claude Code (Claude Fable 5.1)
---
docs/content.zh/docs/managing/job-management.md | 2 +
docs/content/docs/managing/job-management.md | 2 +
.../deployment/AbstractJobReconciler.java | 42 +++++-
.../deployment/ApplicationReconcilerTest.java | 166 +++++++++++++++++++++
.../sessionjob/SessionJobReconcilerTest.java | 45 ++++++
5 files changed, 255 insertions(+), 2 deletions(-)
diff --git a/docs/content.zh/docs/managing/job-management.md
b/docs/content.zh/docs/managing/job-management.md
index ed21c103c..0ded010b4 100644
--- a/docs/content.zh/docs/managing/job-management.md
+++ b/docs/content.zh/docs/managing/job-management.md
@@ -259,6 +259,8 @@ job:
When the `savepointRedeployNonce` changes, the operator redeploys the job to
the savepoint defined in `initialSavepointPath`. The savepoint path must not be
empty.
+A savepoint redeploy requested while the job is
[suspended](#suspending-and-resuming) is applied when the job is resumed. The
job then starts from `initialSavepointPath` regardless of the upgrade mode.
+
{{< hint warning >}}
Rollbacks are not supported after redeployments.
{{< /hint >}}
diff --git a/docs/content/docs/managing/job-management.md
b/docs/content/docs/managing/job-management.md
index ed21c103c..0ded010b4 100644
--- a/docs/content/docs/managing/job-management.md
+++ b/docs/content/docs/managing/job-management.md
@@ -259,6 +259,8 @@ job:
When the `savepointRedeployNonce` changes, the operator redeploys the job to
the savepoint defined in `initialSavepointPath`. The savepoint path must not be
empty.
+A savepoint redeploy requested while the job is
[suspended](#suspending-and-resuming) is applied when the job is resumed. The
job then starts from `initialSavepointPath` regardless of the upgrade mode.
+
{{< hint warning >}}
Rollbacks are not supported after redeployments.
{{< /hint >}}
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 e57aa51b9..9e73258dc 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
@@ -189,8 +189,13 @@ public abstract class AbstractJobReconciler<
}
if (currentJobState == JobState.SUSPENDED && desiredJobState ==
JobState.RUNNING) {
- // We inherit the upgrade mode unless stateless upgrade requested
- if (currentDeploySpec.getJob().getUpgradeMode() !=
UpgradeMode.STATELESS) {
+ // We inherit the upgrade mode unless stateless upgrade requested.
A savepoint the user
+ // explicitly requested through initialSavepointPath is always
restored, regardless of
+ // the upgrade mode of the spec.
+ boolean restoreFromInitialSavepoint =
+ restoringFromInitialSavepoint(resource,
lastReconciledSpec);
+ if (currentDeploySpec.getJob().getUpgradeMode() !=
UpgradeMode.STATELESS
+ || restoreFromInitialSavepoint) {
currentDeploySpec
.getJob()
.setUpgradeMode(lastReconciledSpec.getJob().getUpgradeMode());
@@ -207,10 +212,36 @@ public abstract class AbstractJobReconciler<
lastReconciledSpec.getJob().getUpgradeMode() ==
UpgradeMode.LAST_STATE);
ReconciliationUtils.updateStatusForDeployedSpec(resource,
deployConfig, clock);
+ if (restoreFromInitialSavepoint) {
+ // Savepoint redeploys cannot be rolled back (see
redeployWithSavepoint). That also
+ // applies to a redeploy completed here rather than in
redeployWithSavepoint, i.e.
+ // one requested while suspended, or one whose first
deployment attempt failed.
+
resource.getStatus().getReconciliationStatus().markReconciledSpecAsStable();
+ }
}
return true;
}
+ /**
+ * Checks whether the suspended job is to be restored from the savepoint
the user explicitly
+ * requested through initialSavepointPath, recorded as the upgrade
savepoint by a savepoint
+ * redeploy or the first deployment. Such a savepoint is honoured even for
stateless specs,
+ * otherwise a savepoint redeploy requested while suspended, or one whose
deployment attempt
+ * failed, would be replaced by an empty state restore.
+ *
+ * <p>The check reads initialSavepointPath from the last reconciled spec,
not the current one.
+ * initialSavepointPath is ignored by the spec diff, so editing or
clearing it without a new
+ * savepointRedeployNonce is not a new operation and must not change which
savepoint the
+ * recorded redeploy restores from.
+ */
+ private boolean restoringFromInitialSavepoint(CR resource, SPEC
lastReconciledSpec) {
+ var initialSavepointPath =
lastReconciledSpec.getJob().getInitialSavepointPath();
+ return initialSavepointPath != null
+ && lastReconciledSpec.getJob().getUpgradeMode() ==
UpgradeMode.SAVEPOINT
+ && initialSavepointPath.equals(
+
resource.getStatus().getJobStatus().getUpgradeSavepointPath());
+ }
+
protected JobUpgrade getJobUpgrade(FlinkResourceContext<CR> ctx,
Configuration deployConfig)
throws Exception {
var resource = ctx.getResource();
@@ -633,6 +664,13 @@ public abstract class AbstractJobReconciler<
status.getJobStatus().setUpgradeSavepointPath(savepointPath.orElse(null));
if (desiredJobState == JobState.RUNNING) {
+ // We record the target spec into an upgrading state before
deploying, like every
+ // other deployment path. Otherwise a failed deploy attempt leaves
the new
+ // savepointRedeployNonce unrecorded and every subsequent
reconciliation repeats the
+ // cancel + redeploy cycle instead of restoring from the recorded
savepoint.
+ ReconciliationUtils.updateStatusBeforeDeploymentAttempt(resource,
deployConfig, clock);
+ statusRecorder.patchAndCacheStatus(resource,
ctx.getKubernetesClient());
+
deploy(
ctx,
currentDeploySpec,
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 5b6b36422..a2c7c55eb 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
@@ -140,6 +140,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
import static org.junit.jupiter.params.provider.Arguments.arguments;
@@ -1604,6 +1605,171 @@ public class ApplicationReconcilerTest extends
OperatorTestBase {
}
}
+ @ParameterizedTest
+ @EnumSource(UpgradeMode.class)
+ public void testResumeAfterSavepointRedeployWhileSuspended(UpgradeMode
upgradeMode)
+ throws Exception {
+ var deployment = TestUtils.buildApplicationCluster();
+ deployment.getSpec().getJob().setUpgradeMode(upgradeMode);
+
+ reconciler.reconcile(deployment, context);
+ var runningJobs = flinkService.listJobs();
+ verifyAndSetRunningJobsToStatus(deployment, runningJobs);
+
+ // Suspend, then request a redeploy from an explicit savepoint while
suspended
+ var job = deployment.getSpec().getJob();
+ job.setState(JobState.SUSPENDED);
+ reconciler.reconcile(deployment, context);
+ assertEquals(0, flinkService.getRunningCount());
+
+ job.setInitialSavepointPath("sp-t1");
+ job.setSavepointRedeployNonce(1L);
+ reconciler.reconcile(deployment, context);
+ assertEquals(0, flinkService.getRunningCount());
+ assertEquals("sp-t1",
deployment.getStatus().getJobStatus().getUpgradeSavepointPath());
+
+ // Resuming must start from the requested savepoint regardless of the
upgrade mode
+ job.setState(JobState.RUNNING);
+ reconciler.reconcile(deployment, context);
+ assertEquals(1, runningJobs.size());
+ assertEquals("sp-t1", runningJobs.get(0).f0);
+ }
+
+ @ParameterizedTest
+ @EnumSource(UpgradeMode.class)
+ public void testSavepointRedeployRecoversFromFailedDeployment(UpgradeMode
upgradeMode)
+ throws Exception {
+ var deployment = TestUtils.buildApplicationCluster();
+ deployment.getSpec().getJob().setUpgradeMode(upgradeMode);
+
+ reconciler.reconcile(deployment, context);
+ var runningJobs = flinkService.listJobs();
+ verifyAndSetRunningJobsToStatus(deployment, runningJobs);
+
+ var job = deployment.getSpec().getJob();
+ job.setInitialSavepointPath("sp-t1");
+ job.setSavepointRedeployNonce(1L);
+
+ // The running job is cancelled but the deployment attempt fails
+ flinkService.setDeployFailure(true);
+ assertThrows(Exception.class, () -> reconciler.reconcile(deployment,
context));
+ statusRecorder.updateStatusFromCache(deployment);
+ assertTrue(runningJobs.isEmpty());
+
+ // The target spec must be recorded before the deployment attempt,
otherwise the next
+ // reconciliation would detect the savepointRedeployNonce change again
and repeat the
+ // cancel + redeploy cycle forever
+ var status = deployment.getStatus();
+ assertEquals(ReconciliationState.UPGRADING,
status.getReconciliationStatus().getState());
+ var lastReconciledJob =
+
status.getReconciliationStatus().deserializeLastReconciledSpec().getJob();
+ assertEquals(1L, lastReconciledJob.getSavepointRedeployNonce());
+ assertEquals(JobState.SUSPENDED, lastReconciledJob.getState());
+ assertEquals(UpgradeMode.SAVEPOINT,
lastReconciledJob.getUpgradeMode());
+ assertEquals("sp-t1", status.getJobStatus().getUpgradeSavepointPath());
+
+ // The next reconciliation restores from the recorded savepoint
+ flinkService.setDeployFailure(false);
+ reconciler.reconcile(deployment, context);
+ assertEquals(1, runningJobs.size());
+ assertEquals("sp-t1", runningJobs.get(0).f0);
+ assertEquals(ReconciliationState.DEPLOYED,
status.getReconciliationStatus().getState());
+ lastReconciledJob =
+
status.getReconciliationStatus().deserializeLastReconciledSpec().getJob();
+ assertEquals(1L, lastReconciledJob.getSavepointRedeployNonce());
+ assertEquals(JobState.RUNNING, lastReconciledJob.getState());
+ // Like a redeploy that succeeds on the first attempt, a retried one
cannot be rolled back
+
assertTrue(status.getReconciliationStatus().isLastReconciledSpecStable());
+ }
+
+ @Test
+ public void testSavepointRedeployRetryIsNotRolledBack() throws Exception {
+ var deployment = TestUtils.buildApplicationCluster();
+ deployment.getSpec().getJob().setUpgradeMode(UpgradeMode.SAVEPOINT);
+ offsetReconcilerClock(deployment, Duration.ZERO);
+ deployment
+ .getSpec()
+ .getFlinkConfiguration()
+ .putAllFrom(
+ Map.of(
+
KubernetesOperatorConfigOptions.DEPLOYMENT_ROLLBACK_ENABLED.key(),
+ "true",
+
KubernetesOperatorConfigOptions.DEPLOYMENT_READINESS_TIMEOUT.key(),
+ "10s",
+ KubernetesOperatorConfigOptions
+
.OPERATOR_JOB_UPGRADE_LAST_STATE_FALLBACK_ENABLED
+ .key(),
+ "false"));
+
+ // Initial deployment, mark as stable
+ reconciler.reconcile(deployment, context);
+ var runningJobs = flinkService.listJobs();
+ verifyAndSetRunningJobsToStatus(deployment, runningJobs);
+
deployment.getStatus().getReconciliationStatus().markReconciledSpecAsStable();
+
+ // Savepoint redeploy whose first deployment attempt fails
+ var job = deployment.getSpec().getJob();
+ job.setInitialSavepointPath("sp-t1");
+ job.setSavepointRedeployNonce(1L);
+ flinkService.setDeployFailure(true);
+ assertThrows(Exception.class, () -> reconciler.reconcile(deployment,
context));
+ statusRecorder.updateStatusFromCache(deployment);
+ assertTrue(runningJobs.isEmpty());
+
+ // The retry restores from the requested savepoint
+ flinkService.setDeployFailure(false);
+ reconciler.reconcile(deployment, context);
+ assertEquals(1, runningJobs.size());
+ assertEquals("sp-t1", runningJobs.get(0).f0);
+
+ // Even if the JobManager is not ready within the readiness timeout,
the redeploy must not
+ // be rolled back to the previously stable spec
+ offsetReconcilerClock(deployment, Duration.ofSeconds(15));
+ flinkService.setHaDataAvailable(false);
+ flinkService.setJobManagerReady(false);
+ reconciler.reconcile(deployment, context);
+
+ var status = deployment.getStatus();
+ assertEquals(ReconciliationState.DEPLOYED,
status.getReconciliationStatus().getState());
+ assertEquals(1, runningJobs.size());
+ assertEquals("sp-t1", runningJobs.get(0).f0);
+ assertEquals(
+ 1L,
+ status.getReconciliationStatus()
+ .deserializeLastStableSpec()
+ .getJob()
+ .getSavepointRedeployNonce());
+ }
+
+ @Test
+ public void
testResumeIgnoresInitialSavepointPathEditAfterRedeployWhileSuspended()
+ throws Exception {
+ var deployment = TestUtils.buildApplicationCluster();
+ deployment.getSpec().getJob().setUpgradeMode(UpgradeMode.STATELESS);
+
+ reconciler.reconcile(deployment, context);
+ var runningJobs = flinkService.listJobs();
+ verifyAndSetRunningJobsToStatus(deployment, runningJobs);
+
+ // Suspend, then request a savepoint redeploy
+ var job = deployment.getSpec().getJob();
+ job.setState(JobState.SUSPENDED);
+ reconciler.reconcile(deployment, context);
+ job.setInitialSavepointPath("sp-t1");
+ job.setSavepointRedeployNonce(1L);
+ reconciler.reconcile(deployment, context);
+ assertEquals(0, flinkService.getRunningCount());
+ assertEquals("sp-t1",
deployment.getStatus().getJobStatus().getUpgradeSavepointPath());
+
+ // initialSavepointPath is ignored by the spec diff: clearing it
without a new nonce is not
+ // a new operation, so the job still resumes from the recorded
redeploy's savepoint
+ job.setInitialSavepointPath(null);
+ job.setState(JobState.RUNNING);
+ reconciler.reconcile(deployment, context);
+ assertEquals(1, runningJobs.size());
+ assertEquals("sp-t1", runningJobs.get(0).f0);
+ }
+
private void verifySavepointRedeploy(
FlinkDeployment deployment,
List<Tuple3<String, JobStatusMessage, Configuration>> runningJobs,
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconcilerTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconcilerTest.java
index 12e4550cf..616d527e8 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconcilerTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobReconcilerTest.java
@@ -1127,4 +1127,49 @@ public class SessionJobReconcilerTest extends
OperatorTestBase {
"s3://bucket/savepoint-explicit",
verifyAndReturnTheSubmittedJob(sessionJob,
flinkService.listJobs()).f0);
}
+
+ @Test
+ public void testSavepointRedeployRecoversFromFailedDeployment() throws
Exception {
+ var readyCtx =
TestUtils.createContextWithReadyFlinkDeployment(kubernetesClient);
+ FlinkSessionJob sessionJob = TestUtils.buildSessionJob();
+
+ reconciler.reconcile(sessionJob, readyCtx);
+ verifyAndSetRunningJobsToStatus(
+ sessionJob, JobState.RUNNING, RECONCILING, null,
flinkService.listJobs());
+
+
sessionJob.getSpec().getJob().setInitialSavepointPath("s3://bucket/savepoint-explicit");
+ sessionJob.getSpec().getJob().setSavepointRedeployNonce(1L);
+
+ // The running job is cancelled but the deployment attempt fails
+ flinkService.setDeployFailure(true);
+ assertThrows(Exception.class, () -> reconciler.reconcile(sessionJob,
readyCtx));
+ statusRecorder.updateStatusFromCache(sessionJob);
+
+ // The target spec must be recorded before the deployment attempt,
otherwise the next
+ // reconciliation would detect the savepointRedeployNonce change again
and repeat the
+ // cancel + redeploy cycle forever
+ var status = sessionJob.getStatus();
+ assertEquals(ReconciliationState.UPGRADING,
status.getReconciliationStatus().getState());
+ var lastReconciledJob =
+
status.getReconciliationStatus().deserializeLastReconciledSpec().getJob();
+ assertEquals(1L, lastReconciledJob.getSavepointRedeployNonce());
+ assertEquals(JobState.SUSPENDED, lastReconciledJob.getState());
+ assertEquals(UpgradeMode.SAVEPOINT,
lastReconciledJob.getUpgradeMode());
+ assertEquals(
+ "s3://bucket/savepoint-explicit",
status.getJobStatus().getUpgradeSavepointPath());
+
+ // The next reconciliation restores from the recorded savepoint
+ flinkService.setDeployFailure(false);
+ reconciler.reconcile(sessionJob, readyCtx);
+ assertEquals(
+ "s3://bucket/savepoint-explicit",
+ verifyAndReturnTheSubmittedJob(sessionJob,
flinkService.listJobs()).f0);
+ assertEquals(ReconciliationState.DEPLOYED,
status.getReconciliationStatus().getState());
+ lastReconciledJob =
+
status.getReconciliationStatus().deserializeLastReconciledSpec().getJob();
+ assertEquals(1L, lastReconciledJob.getSavepointRedeployNonce());
+ assertEquals(JobState.RUNNING, lastReconciledJob.getState());
+ // Like a redeploy that succeeds on the first attempt, a retried one
cannot be rolled back
+
assertTrue(status.getReconciliationStatus().isLastReconciledSpecStable());
+ }
}
