[FLINK-40162] Handle UpgradeFailureException in FlinkSessionJobController

[milindl]

What is the purpose of the change

The state of the FlinkSessionJob (status.jobStatus.state and status.state) is stuck and not updated in case of an UpgradeFailureException (for instance, if the job ends quickly and the checkpointing interval is set, but state.checkpoints.dir is not configured).

Brief change log

• Moves observer.observe() inside try-catch and add explicit catch for UpgradeFailureException, similar to FlinkDeploymentController.
• triggerErrorEvent now accepts a reason similar to FlinkDeploymentController

Verifying this change

This change added tests and can be verified as follows:

• testJobFinishedWithNonExternallyAddressableCheckpoint: Submits a session job, simulates it finishing with a checkpoint whose external pointer is non-externally-addressable (checkpointing enabled but no state.checkpoints.dir), and verifies the job status correctly reflects FINISHED with the error recorded.
• Manual testing using minikube using the exact JIRA reproducer in the ticket (SleepJob with 5s delay, execution.checkpointing.interval: 2000, no state.checkpoints.dir):
  • FKO 1.15.0: job stuck at RUNNING/STABLE despite Flink REST API showing FINISHED
  • Patched FKO: job correctly transitions to FINISHED/STABLE with error recorded

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 (I think)

Documentation

• Does this pull request introduce a new feature? no