[FLINK-39953] Fix cluster deletion timeout silently swallowed during upgrade

[lucasgameiroborges]

What is the purpose of the change

During an upgrade, ApplicationReconciler.deploy() calls deleteClusterDeployment() to tear down the old cluster and then immediately calls submitApplicationCluster() to start the new one. If the old JobManager Deployment takes longer to disappear than the configured shutdown timeout, deleteBlocking() was silently catching the KubernetesClientTimeoutException inside a broader KubernetesClientException handler and returning normally, so the operator proceeded to create a new Deployment while the old one was still running. This caused port conflicts on the -rest Service or stale endpoint routing.

This change makes the timeout propagate so the reconciler requeues instead of proceeding.
FLINK-39953

Brief change log

• Adds an explicit catch (KubernetesClientTimeoutException) to AbstractFlinkService.deleteBlocking() before the generic KubernetesClientException handler so the timeout propagates to the caller.

• NativeFlinkService.shutdownJobManagersBlocking(): wrap the deleteBlocking() call for the optional scale-to-zero pre-step in its own catch (KubernetesClientTimeoutException) since a timeout there should not block the actual Deployment deletion from proceeding.

Verifying this change

This change added some unit tests ensuring the new behavior. For the AbstractFlinkServiceTest part:

• testBlockingDeploymentDeletion: The existing timeout case now asserts KubernetesClientTimeoutException is thrown.
• testDeleteBlockingTimeoutThrows: Direct test of deleteBlocking() where the mock server always returns the Deployment, asserts the exception is thrown.

Also added tests to NativeFlinkServiceTest and StandaloneFlinkServiceTest ensuring the expection propagates in both cases.

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: yes (AbstractFlinkService.deleteBlocking)

Documentation

• Does this pull request introduce a new feature? no (bug fix)
• If yes, how is the feature documented? n/a

[Dennis-Mircea]

Thanks for opening this PR! The swallowed timeout is a real smell, but the logging is wrong: deleteBlocking logs ERROR and rethrows (double-logging upstream). In one place you are logging a WARN and in other you are logging ERROR, making it inconsistent. Also, the logging massage framing doesn't look to me either.

From a different point of view, this is addressed only for the native operator path, but not also the standalone part.

Still, before proceeding, I'd suggest confirming the diagnosis on the JIRA with a real reproduction first, then: leave logging to the callers, and audit/test the full set of deleteBlocking callers (standalone + clean-up), not just the native upgrade path.

[lucasgameiroborges]

Thank your for taking the time to review my change @Dennis-Mircea ! Answering to your points:

I added the minimum steps to reproduce the issue on the ticket description in JIRA: https://issues.apache.org/jira/browse/FLINK-39953

Regarding double logging:
I removed the error logging from deleteBlocking before rethrow. However, I kept the warning log on the NativeFlinkService side since the scale-down of JM pods to 0 should not block the following deletions. Let me know if you agree.

Regarding Standalone part:
If I understand correctly, there's nothing to patch there. StandaloneFlinkService.deleteClusterInternal calls deleteDeploymentBlocking from the abstract which will throw the timeout exception as per the fix. It calls it twice to issue "mandatory" deletions of JMs and TMs, so no reason to catch and log warnings here. Added 2 unit tests there to ensure the proper behavior on that part.

Happy to address any other concerns that appear.