vsantwana commented on code in PR #1197:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/1197#discussion_r4421436312
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconciler.java:
##########
@@ -160,15 +160,15 @@ public void deploy(
if (savepoint.isPresent()) {
// Savepoint deployment
- deployConfig.set(SavepointConfigOptions.SAVEPOINT_PATH,
savepoint.get());
+ deployConfig.set(SAVEPOINT_PATH, savepoint.get());
Review Comment:
This writes `execution.state-recovery.path`, not `execution.savepoint.path`.
`Configuration.set` writes only the option's main key; the old key is accepted
when reading but never written. `StateRecoveryOptions` only exists from Flink
1.20, and a Flink 1.19 JobManager reads only `execution.savepoint.path`.
So in native mode, a savepoint upgrade of a 1.19 (or older) deployment would
start the job without its state. The dummy path on line 167 would also stop
guarding last-state upgrades against restoring the wrong state. Standalone mode
is fine, because there the path is passed as `--fromSavepoint`.
Keeping `SavepointConfigOptions.SAVEPOINT_PATH` here avoids this: it still
writes the old key, and Flink 1.20 through 2.3 still read it as a deprecated
key. The tests don't catch it because they read the value back through the same
constant.
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilder.java:
##########
@@ -164,17 +164,17 @@ protected static void applyJobConfig(String name,
Configuration conf, JobSpec jo
if (jobSpec.getUpgradeMode() == UpgradeMode.LAST_STATE) {
setDefaultConf(
conf,
- ExecutionCheckpointingOptions.CHECKPOINTING_INTERVAL,
+ CheckpointingOptions.CHECKPOINTING_INTERVAL,
DEFAULT_CHECKPOINTING_INTERVAL);
}
setDefaultConf(
conf,
- ExecutionCheckpointingOptions.EXTERNALIZED_CHECKPOINT,
-
CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
+ CheckpointingOptions.EXTERNALIZED_CHECKPOINT_RETENTION,
+ ExternalizedCheckpointRetention.RETAIN_ON_CANCELLATION);
if (jobSpec.getAllowNonRestoredState() != null) {
conf.set(
- SavepointConfigOptions.SAVEPOINT_IGNORE_UNCLAIMED_STATE,
+ StateRecoveryOptions.SAVEPOINT_IGNORE_UNCLAIMED_STATE,
Review Comment:
Same issue as in `ApplicationReconciler`: this writes
`execution.state-recovery.ignore-unclaimed-state`, which Flink 1.19 and older
don't read, so `allowNonRestoredState: true` in the spec would be ignored on
those versions. `SavepointConfigOptions.SAVEPOINT_IGNORE_UNCLAIMED_STATE` works
on all of them.
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/bluegreen/BlueGreenKubernetesService.java:
##########
@@ -92,7 +96,7 @@ public static void
replaceFlinkBlueGreenDeployment(BlueGreenContext context) {
.getClient()
.resource(context.getBgDeployment())
.inNamespace(namespace)
- .replace();
+ .update();
Review Comment:
This changes behaviour a little. `replace()` retried a conflict with the
latest resourceVersion, while `update()` uses the version on the cached
`bgDeployment` and fails with a 409 if someone edited the resource in the
meantime. I think that's fine (the reconcile retries, and it no longer
overwrites a newer edit), but it's worth a line in the description, since the
PR reads as an API-only change.
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkOperatorConfiguration.java:
##########
@@ -313,20 +313,30 @@ private static LeaderElectionConfiguration
getLeaderElectionConfig(Configuration
return null;
}
- return new LeaderElectionConfiguration(
- conf.getOptional(
-
KubernetesOperatorConfigOptions.OPERATOR_LEADER_ELECTION_LEASE_NAME)
- .orElseThrow(
- () ->
- new IllegalConfigurationException(
- KubernetesOperatorConfigOptions
-
.OPERATOR_LEADER_ELECTION_LEASE_NAME
- .key()
- + " must be defined
when operator leader election is enabled.")),
- null,
-
conf.get(KubernetesOperatorConfigOptions.OPERATOR_LEADER_ELECTION_LEASE_DURATION),
-
conf.get(KubernetesOperatorConfigOptions.OPERATOR_LEADER_ELECTION_RENEW_DEADLINE),
-
conf.get(KubernetesOperatorConfigOptions.OPERATOR_LEADER_ELECTION_RETRY_PERIOD));
+ return LeaderElectionConfigurationBuilder.aLeaderElectionConfiguration(
Review Comment:
Following up on this: the JOSDK fix
(operator-framework/java-operator-sdk#3619) first shipped in 5.6.2, and this
repo is still on 5.5.0. With `.build()`, the operator would keep running after
it loses its lease. Until JOSDK is bumped, `.buildForTest(true)` or keeping the
deprecated constructor for this one call keeps the current behaviour.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
