This is an automated email from the ASF dual-hosted git repository.
gyfora pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git
The following commit(s) were added to refs/heads/main by this push:
new 17723ff [FLINK-28348] Add configurable flag to disable last-state
fallback for savepoint upgrade
17723ff is described below
commit 17723ffc09b6813006044f50165bb6cf15427df7
Author: Matyas Orhidi <[email protected]>
AuthorDate: Mon Jul 4 19:42:26 2022 +0200
[FLINK-28348] Add configurable flag to disable last-state fallback for
savepoint upgrade
---
.../kubernetes_operator_config_configuration.html | 6 +++++
.../config/KubernetesOperatorConfigOptions.java | 7 +++++
.../deployment/ApplicationReconciler.java | 6 ++++-
.../controller/FlinkDeploymentControllerTest.java | 31 +++++++++++++++++++++-
4 files changed, 48 insertions(+), 2 deletions(-)
diff --git
a/docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html
b/docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html
index ec5edc6..cfc1ba9 100644
---
a/docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html
+++
b/docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html
@@ -74,6 +74,12 @@
<td>Boolean</td>
<td>Whether to ignore pending savepoint during job upgrade.</td>
</tr>
+ <tr>
+
<td><h5>kubernetes.operator.job.upgrade.last-state-fallback.enabled</h5></td>
+ <td style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>Enables last-state fallback for savepoint upgrade mode. When
the job is not running thus savepoint cannot be triggered but HA metadata is
available for last state restore the operator can initiate the upgrade process
when the flag is enabled.</td>
+ </tr>
<tr>
<td><h5>kubernetes.operator.observer.progress-check.interval</h5></td>
<td style="word-wrap: break-word;">10 s</td>
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java
index a9f253b..aa6031c 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java
@@ -230,4 +230,11 @@ public class KubernetesOperatorConfigOptions {
.defaultValue(10)
.withDescription(
"Max attempts of automatic reconcile retries on
recoverable errors.");
+
+ public static final ConfigOption<Boolean>
OPERATOR_JOB_UPGRADE_LAST_STATE_FALLBACK_ENABLED =
+
ConfigOptions.key("kubernetes.operator.job.upgrade.last-state-fallback.enabled")
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "Enables last-state fallback for savepoint upgrade
mode. When the job is not running thus savepoint cannot be triggered but HA
metadata is available for last state restore the operator can initiate the
upgrade process when the flag is enabled.");
}
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconciler.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconciler.java
index 688d2d6..2747349 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconciler.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconciler.java
@@ -21,6 +21,7 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.HighAvailabilityOptions;
import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
+import
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
import org.apache.flink.kubernetes.operator.crd.spec.FlinkDeploymentSpec;
import org.apache.flink.kubernetes.operator.crd.spec.UpgradeMode;
@@ -85,7 +86,10 @@ public class ApplicationReconciler
return availableUpgradeMode;
}
- if (FlinkUtils.isKubernetesHAActivated(deployConfig)
+ if (deployConfig.getBoolean(
+ KubernetesOperatorConfigOptions
+
.OPERATOR_JOB_UPGRADE_LAST_STATE_FALLBACK_ENABLED)
+ && FlinkUtils.isKubernetesHAActivated(deployConfig)
&& FlinkUtils.isKubernetesHAActivated(observeConfig)
&& flinkService.isHaMetadataAvailable(deployConfig)) {
LOG.info(
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentControllerTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentControllerTest.java
index c809418..788b770 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentControllerTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentControllerTest.java
@@ -66,6 +66,7 @@ import java.util.Optional;
import java.util.function.BiConsumer;
import java.util.stream.Stream;
+import static
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.OPERATOR_JOB_UPGRADE_LAST_STATE_FALLBACK_ENABLED;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
@@ -562,7 +563,7 @@ public class FlinkDeploymentControllerTest {
FlinkVersion flinkVersion, UpgradeMode upgradeMode) throws
Exception {
var appCluster = TestUtils.buildApplicationCluster(flinkVersion);
appCluster.getSpec().getJob().setUpgradeMode(upgradeMode);
- testUpgradeNotReadyCluster(appCluster);
+ testUpgradeNotReadyCluster(ReconciliationUtils.clone(appCluster));
}
@Test
@@ -742,6 +743,34 @@ public class FlinkDeploymentControllerTest {
testController.reconcile(appCluster, context);
testController.reconcile(appCluster, context);
+ assertEquals(
+ org.apache.flink.api.common.JobStatus.RUNNING.name(),
+ appCluster.getStatus().getJobStatus().getState());
+ assertEquals(
+ JobManagerDeploymentStatus.READY,
+ appCluster.getStatus().getJobManagerDeploymentStatus());
+
+ // triggering upgrade with no last-state fallback on non-healthy app
+ flinkService.setPortReady(false);
+ appCluster
+ .getSpec()
+ .getFlinkConfiguration()
+ .put(OPERATOR_JOB_UPGRADE_LAST_STATE_FALLBACK_ENABLED.key(),
"false");
+
appCluster.getSpec().setServiceAccount(appCluster.getSpec().getServiceAccount()
+ "-5");
+ // not upgrading the cluster with no last-state fallback
+ testController.reconcile(appCluster, context);
+ assertNotEquals(
+ appCluster.getSpec(),
+
appCluster.getStatus().getReconciliationStatus().deserializeLastReconciledSpec());
+
+ // once the job is ready however the upgrade continues
+ flinkService.setPortReady(true);
+ testController.reconcile(appCluster, context);
+ testController.reconcile(appCluster, context);
+ testController.reconcile(appCluster, context);
+ testController.reconcile(appCluster, context);
+ testController.reconcile(appCluster, context);
+
assertEquals(
org.apache.flink.api.common.JobStatus.RUNNING.name(),
appCluster.getStatus().getJobStatus().getState());
