gyfora commented on code in PR #622:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/622#discussion_r1242139767
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractJobReconciler.java:
##########
@@ -113,9 +113,22 @@ protected boolean
reconcileSpecChange(FlinkResourceContext<CR> ctx, Configuratio
EventRecorder.Reason.Suspended,
EventRecorder.Component.JobManagerDeployment,
MSG_SUSPENDED);
+
+ UpgradeMode upgradeMode =
availableUpgradeMode.getUpgradeMode().get();
+
// We must record the upgrade mode used to the status later
-
currentDeploySpec.getJob().setUpgradeMode(availableUpgradeMode.getUpgradeMode().get());
- cancelJob(ctx, availableUpgradeMode.getUpgradeMode().get());
+ currentDeploySpec.getJob().setUpgradeMode(upgradeMode);
+
+ // We must use LAST_STATE mode when rolling back from
+ // SAVEPOINT. But we don't want to set upgrade mode to LAST_STATE
+ // as we will rely on SAVEPOINT restoreJob mechanism
+ if (upgradeMode == UpgradeMode.SAVEPOINT
+ && status.getReconciliationStatus().getState()
+ == ReconciliationState.ROLLING_BACK) {
+ upgradeMode = UpgradeMode.LAST_STATE;
+ }
Review Comment:
Can you please explain what's going on here? I don't really understand why
we change from savepoint -> last-state for the cancel operation specifically.
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java:
##########
@@ -138,14 +138,32 @@ public void reconcile(FlinkResourceContext<CR> ctx)
throws Exception {
DiffType.IGNORE != diffType
|| reconciliationStatus.getState() ==
ReconciliationState.UPGRADING;
+ if (!specChanged && reconciliationStatus.getState() ==
ReconciliationState.ROLLING_BACK) {
+ // Rely on the last stable spec if rolling back and no change in
the spec
+
cr.setSpec(cr.getStatus().getReconciliationStatus().deserializeLastStableSpec());
+ specChanged = true;
+ } else if (specChanged
+ && reconciliationStatus.getState() ==
ReconciliationState.ROLLING_BACK) {
+ // Spec has changed while rolling back we should apply new spec
and move to upgrading
+ // state
+ // Don't take in account changes on job.state as it could be
overriden to running if the
+ // current spec is not valid
+
lastReconciledSpec.getJob().setState(currentDeploySpec.getJob().getState());
+ var specDiffRollingBack =
+ new ReflectiveDiffBuilder<>(
+ ctx.getDeploymentMode(),
lastReconciledSpec, currentDeploySpec)
+ .build();
+ if (DiffType.IGNORE != specDiffRollingBack.getType()) {
+ reconciliationStatus.setState(ReconciliationState.UPGRADING);
+ }
+ }
Review Comment:
It would be great to encapsulate this into a method that mutates the
CR/status fields and returns the value of specChanged variable that we should
set afterwards.
It could be something like:
```
if (reconciliationStatus.getState() == ReconciliationState.ROLLING_BACK) {
specChanged = prepareCrForRollback(...)
}
```
--
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]
