[hotfix] Replace deprecated Flink, JOSDK, and fabric8 APIs

[Dennis-Mircea]

What is the purpose of the change

The purpose of this PR is to replace the external deprecated Flink core, JOSDK, and fabric8 APIs.

Brief change log

Flink-related:

• Replaced the deprecated typed Configuration accessors with the generic ones.
• Replaced the deprecated ExecutionCheckpointingOptions with CheckpointingOptions.
• Replaced the deprecated SavepointConfigOptions with StateRecoveryOptions .

JOSDK-related:

• Replaced the deprecated installShutdownHook method.
• Replaced the deprecated LeaderElectionConfiguration with LeaderElectionConfigurationBuilder.

Fabric8-related:

• Replaced the deprecated replace() with update().
• Replaced the deprecated createOrReplace() with createOr(NonDeletingOperation::update) or createOr(r -> r.unlock().update()) or forEach(Resource::update) based on the case.

Verifying this change

This change can be verified with the existing tests that still pass with this update.

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)

[spuru9]

Did a small pass over the change.

[spuru9]

In JOSDK 5.5.0, LeaderElectionConfigurationBuilder.build() is implemented as:

    public LeaderElectionConfiguration build() {
        return buildForTest(false);
    }

This hardcodes exitOnStopLeading = false (the builder's Javadoc notes false should only be used for tests).

Previously, the deprecated constructor new LeaderElectionConfiguration(...) set exitOnStopLeading = true. When exitOnStopLeading is true, if the operator loses its leadership lease, LeaderElectionManager.stopLeading() executes System.exit(1) so Kubernetes restarts the pod and allows a standby replica to take over cleanly.

With .build(), exitOnStopLeading becomes false, so stopLeading() will merely log "Stopped leading, configured not to exit" and keep running without exiting. This can cause split-brain or stuck reconciliation threads during HA failovers.

Please use .buildForTest(true) here so that production HA failover semantics (System.exit(1) on lost lease) are preserved.

[csviri]

Yes, this seems to be a bug in JOSDK, will create a patch for that. Will be part of next patch release version.

[csviri]

operator-framework/java-operator-sdk#3619

[Dennis-Mircea]

Should we wait for the new JOSDK patch to be adopted as part of the operator or is it fine to go with the workaround that @spuru9 suggested? cc @gyfora

[spuru9]

Given we wont be shipping this in this release and I see a PR for the JOSDK patch, it should be fine. We can add a note in this PR or the other that this is changing the following behavior.

[gyfora]

I leave this decision to @csviri , he knows the expected timeline for the fix. We could also just avoid touching this part until the josdk patch is available

[spuru9]

configMaps contains all HA ConfigMaps for the cluster (e.g. dispatch leader, JM leader, etc.). Calling .resources().forEach(Resource::update) sends an HTTP PUT update for every single ConfigMap in the list, even those that were untouched.

1. It issues unnecessary `PUT` requests for unmodified ConfigMaps.
2. If any untouched ConfigMap (such as an active leader lease ConfigMap) was concurrently modified in the cluster, its `resourceVersion` will be stale and `Resource::update` will throw a `409 Conflict` exception, aborting the cleanup.

Suggestion: collect only the modified ConfigMaps in the loop above and update only those:

            List<ConfigMap> modifiedConfigMaps = new ArrayList<>();
            for (ConfigMap configMap : configMaps.getItems()) {
                if (configMap.getData() == null || configMap.getData().isEmpty()) {
                    continue;
                }
                if (configMap.getData().entrySet().removeIf(FlinkUtils::isJobGraphKey)) {
                    modifiedConfigMaps.add(configMap);
                    LOG.info("Job graph in ConfigMap {} is deleted", configMap.getMetadata().getName());
                }
            }
            modifiedConfigMaps.forEach(cm -> kubernetesClient.resource(cm).inNamespace(namespace).update());

[Dennis-Mircea]

Good suggestion, I've added it.

[gyfora]

why not NonDeletingOperation::update?

[Dennis-Mircea]

The thing is that, having just update in place, this will perform an optimistic-lock PUT, and if multiple concurrent updates happen at the same time, it will result in 409 and it will fail the reconciliation cycle.

The old createOrReplace was a forced replace, which is what we want for best-effort events, and unlock() was meant to keep that.

Looking again at this point from a different perspective, the unlock() only runs on the conflict branch, after the create returns, but create can fail with a 500 when a POST is executed with the same resourceVersion (see https://github.com/kubernetes/kubernetes/blob/master/staging/src/k8s.io/apiserver/pkg/storage/errors.go#L30).

To resolve this, I used the fabric8 migration FAQ recommendation presented here: https://github.com/fabric8io/kubernetes-client/blob/main/doc/FAQ.md#alternatives-to-createorreplace-and-replace. I think we should all good with this approach.

[spuru9]

LGTM % CI passes