[FLINK-27572][FLINK-27594] Ensure HA metadata is present before restoring job with last state

[gyfora]

This PR contains the following improvements/fixes for a number of loosely connected blocker issues:

1. Eliminate bug that causes 0 delay infinite reconcile loop with the UpdateControl management logic
2. Ensure HA metadata is available when we are relying on it during stateful upgrades
3. Make JM deployment recovery condititional on availablity of HA metadata
4. Trigger fatal error when upgrade progress is stuck
5. Clean up and improve stateful upgrade conditions with added detailed debug logging
6. Greately simplify code around suspend/cancellation and restore operation in the reconciler
7. Make sure finished jobs are marked as stable to avoid rollback loop for short jobs

[gyfora]

cc @Aitozi @tweise @wangyang0918 @morhidi

This is a larger change that fixes a couple outstanding critical issues and also aims to make the whole flow a bit simpler, more robust and easier to understand. I would appreciate your review/feedback :)

[wangyang0918]

Great job for dealing with these important blocker issues in one shot. I have carefully gone though this PR and do not have major comments. I will play this PR with some more manual tests today and then share the feedback.

4. Trigger fatal error when upgrade progress is stuck

BTW, I do not fully understand why the upgrade progress could be stuck and how you fix it.

[wangyang0918]

It could be done in another ticket. But we might need to document that how to do the manual restore. It is important for the release 1.14 and 1.13.

[gyfora]

I have a ticket open for these docs, I will work on that next week to have it for the release.

By the way this is the exact situation that answers your question regarding the "stuck" upgrade progress. This is a situation where the JM deployment is missing, and also HA is not available to get last state info.

We cannot deal with it, that's why we throw an error.

[wangyang0918]

Thanks for the hint.

[wangyang0918]

I have done some manual tests with both 1.14 and 1.15 versions. It works smoothly and I believe the Flink deployment is pretty more robust now. 👍🏻

• Invalid image and rollback
• Cancel/Fail Flink and verify last-state upgrade. Expect fatal error in 1.14 and successful recovery for 1.15.
• Recover missing JobManager deployment
• Checking the status and operator logs

I found some small issues when testing.

1. It is unnecessary to trigger an upgrade when the spec is exactly same with stable spec. For example, the invalid image is reverted.
2. We are generating too many ERROR events for FlinkDeployment. Each reconciliation will append a new one.
   

[gyfora]

Thanks @wangyang0918 for the testing. I will improve the rollback -> upgrade scenario that you encountered.

The events should also be fixed after merging #213 will open a separate blocker ticket for that.