[FLINK-40374] Fix redeploy-from-savepoint for suspended jobs

[dalelane]

Redeploying a suspended FlinkSessionJob from an explicit savepoint (state SUSPENDED -> RUNNING via savepointRedeployNonce) always tried to cancel the job first, even though the suspend clears the jobId. FlinkDeployment guards this with isJobCancellable() but the session job path was missing the same guard.

What is the purpose of the change

For a FlinkSessionJob, if a job is suspended with a savepoint, and the user then tries to explicitly redeploy it from a specific savepoint path (by setting spec.job.state=RUNNING, spec.job.initialSavepointPath, and bumping spec.job.savepointRedeployNonce), the reconciler would fail.

Brief change log

• Created a unit test to recreate the issue
• Copied the guard used for FlinkDeployment to FlinkSessionJob.

Verifying this change

This change added a unit test. It can also be verified manually by suspending a job on a session cluster and then resuming from a savepoint.

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changes to the CustomResourceDescriptors: no
• Core observer or reconciler logic that is regularly executed: no

Documentation

• Does this pull request introduce a new feature? no
• If yes, how is the feature documented? not applicable

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

[milindl]

Makes sense, and the PR looks good. Tests also look good.

[Dennis-Mircea]

Thanks for opening the PR! The diagnosis is correct, but I don't think a state check is the right fix.

The two paths differ in what actually stops the job:

• For a FlinkDeployment a stateless suspend deletes the JobManager (SuspendMode.deleteCluster()), so the teardown is what stops the job and the cancel call is only an optimisation. Whether it fails, is skipped, or is still in flight makes no difference, which is why that branch can swallow the exception and drop the return value.
• For a FlinkSessionJob the cluster is shared and survives the suspend, so the cancel is the only thing that stops the job. It has to succeed, and an in-flight cancellation has to be awaited.

I'd say no state check is needed here at all, because cancelJobOrError already absorbs a missing job and a terminated one. With ignoreMissing=true, which is what STATELESS passes, both return true and the caller proceeds. The only thing that actually breaks is JobID.fromHexString(null), which runs before any of that.

I suggest handling the null where it occurs instead, which also matches the documented contract since a null jobId is "already missing":

var jobIdString = status.getJobStatus().getJobId();
if (jobIdString == null) {
    if (ignoreMissing) {
        LOG.info("Job already missing");
        return true;
    }
    throw new UpgradeFailureException(
            "Cannot find job when trying to cancel", EventRecorder.Reason.CleanupFailed.name());
}
var jobID = JobID.fromHexString(jobIdString);

[dalelane]

@Dennis-Mircea thanks so much for the fast review - your approach makes more sense, I've updated the branch with it now