vsantwana commented on code in PR #1239:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/1239#discussion_r4427394239
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/SnapshotObserver.java:
##########
@@ -171,6 +172,17 @@ private void
observeTriggeredSavepoint(FlinkResourceContext<CR> ctx, String jobI
// In case of periodic and manual savepoint, we still use lastSavepoint
savepointInfo.updateLastSavepoint(savepoint);
+
+ // Restores prefer the upgrade savepoint over lastSavepoint, so it has
to follow the newest
+ // savepoint, otherwise a restore rolls the job back to the state of
the last upgrade.
+ // Without one restores already use lastSavepoint, and the last-state
marker means the
+ // latest state is only available in the HA metadata.
+ var jobStatus = resource.getStatus().getJobStatus();
+ var upgradeSavepointPath = jobStatus.getUpgradeSavepointPath();
+ if (upgradeSavepointPath != null
+ &&
!AbstractJobReconciler.LAST_STATE_DUMMY_SP_PATH.equals(upgradeSavepointPath)) {
Review Comment:
The comment says the last-state marker means the latest state is only in the
HA metadata. That stops being true once the HA data is lost, which is the case
this PR is about. Take a savepoint-mode job whose last upgrade fell back to
last-state (`job.upgrade.last-state-fallback.enabled` is on by default). It
keeps the marker. If it then fails without HA data, the next upgrade stops with
"manual restore required", although a newer periodic savepoint exists. Refusing
is safer than restoring old state, so I'm not asking for a change, but is this
intended? If so, it would help to say so in the comment.
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/SnapshotObserver.java:
##########
@@ -171,6 +172,17 @@ private void
observeTriggeredSavepoint(FlinkResourceContext<CR> ctx, String jobI
// In case of periodic and manual savepoint, we still use lastSavepoint
savepointInfo.updateLastSavepoint(savepoint);
+
+ // Restores prefer the upgrade savepoint over lastSavepoint, so it has
to follow the newest
+ // savepoint, otherwise a restore rolls the job back to the state of
the last upgrade.
+ // Without one restores already use lastSavepoint, and the last-state
marker means the
+ // latest state is only available in the HA metadata.
+ var jobStatus = resource.getStatus().getJobStatus();
+ var upgradeSavepointPath = jobStatus.getUpgradeSavepointPath();
+ if (upgradeSavepointPath != null
+ &&
!AbstractJobReconciler.LAST_STATE_DUMMY_SP_PATH.equals(upgradeSavepointPath)) {
+ jobStatus.setUpgradeSavepointPath(savepoint.getLocation());
Review Comment:
This only covers legacy mode. With
`kubernetes.operator.snapshot.resource.enabled: true`, which is the default,
periodic and manual savepoints are FlinkStateSnapshot resources. They never
reach `observeTriggeredSavepoint`, and nothing else moves
`upgradeSavepointPath` when one completes. So most users keep the same failure:
a FAILED job without HA data is redeployed from the last upgrade savepoint.
With the default `savepoint.dispose-on-delete: false`, the file still exists,
so the job silently restarts from the old state instead of failing.
I checked this with a copy of
`testFailedJobRestoredFromLatestSavepointAfterUpgrade` that leaves snapshot
resources on and adds a completed periodic savepoint resource. The upgrade of
the failed job redeployed from `savepoint_0`.
The choice of restore savepoint is made in `restoreJob`, which has preferred
`upgradeSavepointPath` since 1.10 (FLINK-35265); in 1.9 it used
`lastSavepoint`. Could the fix go there, or could the field also be moved when
a savepoint resource completes, so both modes are covered? If you would rather
keep this PR to legacy mode, could you open a follow-up ticket and mention it
in the description?
##########
flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentControllerTest.java:
##########
@@ -438,6 +439,61 @@ public void
verifyUpgradeFromSavepointLegacyMode(FlinkVersion flinkVersion) thro
assertEquals(0, jobs.size());
}
+ @Test
+ public void testFailedJobRestoredFromLatestSavepointAfterUpgrade() throws
Exception {
Review Comment:
This covers an upgrade of the failed job. The Jira also asks about
`kubernetes.operator.job.restart.failed`, which goes through `resubmitJob` and
should now restore from the newest savepoint too. Could you add that case, and
one for `FlinkSessionJob`, which reaches the same observer code?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
