Saurabh Kanawade created FLINK-41033:
----------------------------------------
Summary: Savepoint history cleanup disposes the savepoint
referenced by status.jobStatus.upgradeSavepointPath, causing restore failures
(FileNotFoundException)
Key: FLINK-41033
URL: https://issues.apache.org/jira/browse/FLINK-41033
Project: Flink
Issue Type: Bug
Components: Kubernetes Operator
Affects Versions: kubernetes-operator-1.13.0
Reporter: Saurabh Kanawade
Flink Kubernetes Operator 1.13.0 (upgraded from 1.8.0), Flink 1.18.1,
application mode (FlinkDeployment), upgradeMode: savepoint, Kubernetes HA
(ConfigMap based), adaptive scheduler, FlinkStateSnapshot CRs not used (legacy
savepointHistory), AWS EKS, checkpoints/savepoints/HA on S3, state backend
We recently upgraded the Flink Kubernetes Operator from 1.8.0 to 1.13.0. Since
then, one of our production jobs went down for about 9.5 hours and could not
recover by itself. The restore failed with {{FileNotFoundException}} on a
savepoint that the operator's own savepoint-history cleanup had deleted 28 days
earlier. The operator still listed that savepoint as
{{{}status.jobStatus.upgradeSavepointPath{}}}. We could only recover by
manually redeploying with {{initialSavepointPath}} pointing at a newer
savepoint. The day after the manual fix, the operator deleted that savepoint
too, so the job is again in the same position.
We have not found a configuration that keeps periodic savepoint cleanup and
still guarantees the job can be recovered. We would appreciate guidance from
the community. Our questions are at the end.
{code:java}
||Time||Event||
|09 Sep 22:13:52|Spec change → savepoint upgrade. Job {{2804d6…}} suspended
with {{savepoint-2804d6-c6cd134a569d}}.|
|09 Sep 22:15:36|New job {{61c71cfe5ef7d6b55c3586c04bd1b68f}} starts from that
savepoint; the operator sets {{upgradeSavepointPath}} to it.|
|10 Sep 19:12:25|The operator ({{SnapshotObserver}}) disposes
{{savepoint-2804d6-c6cd134a569d}}, about 21h after it was written. It dropped
out of the 96-entry history.|
|15 Sep … 3 Oct|Status still shows {{upgradeSavepointPath:
…/savepoint-2804d6-c6cd134a569d}}. No event or warning that the file is gone.|
|24 Sep 22:23:51|JobManager restart; the job recovers normally from an HA
checkpoint (same job ID, new start time).|
|07 Oct ≈22:41|Last periodic savepoint taken:
{{savepoint-61c71c-a40dbf62613a}}.|
|07 Oct 22:50–22:58|Node disruption in the cluster. Both JobManagers,
TaskManagers and both operator pods are rescheduled. 5 other Flink jobs hit by
the same event recover within minutes.|
|07 Oct 22:51:24|Our job goes to RESTARTING on the old leader JM after losing a
TaskManager.|
|07 Oct 22:52:48|New JM leader takes over. It finds *no completed checkpoint in
the HA store* and falls back to {{execution.savepoint.path}} (the 9 Sep
savepoint).|
|07 Oct 22:53:43|*Restore fails: FileNotFoundException. Job goes to FAILED.*
{{jobManagerDeploymentStatus}} stays READY.|
|08 Oct 08:08:23|A spec change makes the operator redeploy the FAILED job ("Job
is in terminal state, ready for upgrade from observed latest state").|
|08 Oct 08:09:52|*Fails again with the same FileNotFoundException.* The
operator restored from {{upgradeSavepointPath}}, not from the newer 22:41
savepoint recorded in {{savepointInfo.lastSavepoint}}.|
|08 Oct 08:23:36|Manual {{initialSavepointPath =
…/savepoint-61c71c-a40dbf62613a}} plus {{savepointRedeployNonce}} bump. Job
RUNNING at 08:24:53.|
|08 Oct 22:42:16|The operator disposes {{savepoint-61c71c-a40dbf62613a}}, the
savepoint the new job was just started from. It is still the
{{upgradeSavepointPath}}.|
{code}
h3. Failure logs
1) Savepoint created on upgrade, then disposed by the operator:
{noformat}
2026-09-09 22:13:52 AbstractFlinkService Job successfully suspended with
savepoint
s3://<bucket>/<namespace>/<job>/savepoint/savepoint-2804d6-c6cd134a569d
2026-09-10 19:12:25 SnapshotObserver Disposing savepoint
s3://<bucket>/<namespace>/<job>/savepoint/savepoint-2804d6-c6cd134a569d
{noformat}
2) Operator status 23 days later, still pointing to the deleted savepoint:
{noformat}
2026-10-03 09:25:33 (status logged by operator)
"jobId":"61c71cfe5ef7d6b55c3586c04bd1b68f","state":"RUNNING","startTime":"1790288631452",
"upgradeSavepointPath":"s3://<bucket>/<namespace>/<job>/savepoint/savepoint-2804d6-c6cd134a569d"
{noformat}
3) JobManager failover: restore failure:
{noformat}
2026-10-07 22:51:24 Job restarting after TaskManager loss:
FlinkExpectedException: The TaskExecutor is shutting down
2026-10-07 22:52:39 Operator: Exception while getting job status (JM
deployment DEPLOYED_NOT_READY)
2026-10-07 22:52:48 New JobManager leader; job RECONCILING -> INITIALIZING
2026-10-07 22:53:43 Event[Job] | Warning | ERROR |
java.util.concurrent.CompletionException:
java.io.FileNotFoundException: Cannot find checkpoint or savepoint
file/directory
's3://<bucket>/<namespace>/<job>/savepoint/savepoint-2804d6-c6cd134a569d' on
file system 's3'.
2026-10-07 22:53:43 Event[Job] | Info | JOBSTATUSCHANGED | Job status changed
from INITIALIZING to FAILED
{noformat}
4) Operator redeploy of the FAILED job: same deleted path:
{noformat}
2026-10-08 08:08:23 Job is in terminal state, ready for upgrade from observed
latest state
2026-10-08 08:09:52 Event[Job] | Warning | ERROR |
java.io.FileNotFoundException: Cannot find checkpoint or savepoint
file/directory
's3://<bucket>/<namespace>/<job>/savepoint/savepoint-2804d6-c6cd134a569d' on
file system 's3'.
{noformat}
5) Manual recovery, and the new restore savepoint deleted again:
{noformat}
Here after manually updated the initial savepoint path & redeployNounce so we
have recovered the application.{noformat}
h3. Questions for the community
# Is it expected that the operator's savepoint-history cleanup disposes the
savepoint currently stored in {{{}status.jobStatus.upgradeSavepointPath{}}},
which is also the running cluster's {{{}execution.savepoint.path{}}}? If so,
what retention setup do you recommend so that a job can always be recovered?
# For a job in terminal FAILED state with no observable checkpoint, why does
the operator restore from {{upgradeSavepointPath}} instead of the newer
{{{}savepointInfo.lastSavepoint{}}}? Is there a configuration that makes it use
the latest completed savepoint?
# Is there a supported way, such as
{{{}kubernetes.operator.job.restart.failed{}}}, to automatically recover a
FAILED job in this situation without manual {{initialSavepointPath}} changes?
Would it also hit the deleted path?
# Could the empty HA checkpoint store after the JobManager failover at 22:52
be explained by FLINK-34518 (Flink 1.18.1 plus adaptive scheduler, JM lost
while the job was RESTARTING)? Are there other known causes? Is upgrading to
Flink 1.18.2+ the right fix?
# Would enabling FlinkStateSnapshot CRs change this behaviour, or are UPGRADE
snapshots cleaned up the same way there?
# Would you recommend {{upgradeMode: last-state}} instead of {{savepoint}} for
this kind of setup?
# If this is considered a bug, would the maintainers accept a fix that skips
disposing the savepoint referenced by {{{}upgradeSavepointPath{}}}? We are
happy to contribute a PR.
Thank you for any guidance.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
