You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gy...@apache.org on 2022/05/19 10:25:27 UTC
[flink-kubernetes-operator] branch release-1.0 updated: [hotfix][FLINK-27572] Harden HA meta checking logic
This is an automated email from the ASF dual-hosted git repository.
gyfora pushed a commit to branch release-1.0
in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git
The following commit(s) were added to refs/heads/release-1.0 by this push:
new 5e0d7f9 [hotfix][FLINK-27572] Harden HA meta checking logic
5e0d7f9 is described below
commit 5e0d7f9d17df4c7f1510a4fea6ef84f648996d41
Author: Gyula Fora <g_...@apple.com>
AuthorDate: Thu May 19 06:43:22 2022 +0200
[hotfix][FLINK-27572] Harden HA meta checking logic
---
.../deployment/ApplicationReconciler.java | 54 ++++++++++++----------
.../kubernetes/operator/service/FlinkService.java | 8 ++--
.../kubernetes/operator/utils/FlinkUtils.java | 1 +
.../kubernetes/operator/TestingFlinkService.java | 7 ++-
.../controller/FlinkDeploymentControllerTest.java | 8 ++++
5 files changed, 48 insertions(+), 30 deletions(-)
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 77b9cc5..194a545 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
@@ -100,18 +100,13 @@ public class ApplicationReconciler extends AbstractDeploymentReconciler {
return;
}
+ Configuration observeConfig = configManager.getObserveConfig(flinkApp);
boolean specChanged = !currentDeploySpec.equals(lastReconciledSpec);
if (specChanged) {
if (newSpecIsAlreadyDeployed(flinkApp)) {
return;
}
LOG.debug("Detected spec change, starting upgrade process.");
- Optional<UpgradeMode> availableUpgradeMode = getAvailableUpgradeMode(flinkApp);
- if (availableUpgradeMode.isEmpty()) {
- return;
- }
- UpgradeMode upgradeMode = availableUpgradeMode.get();
-
JobState currentJobState = lastReconciledSpec.getJob().getState();
JobState desiredJobState = desiredJobSpec.getState();
JobState stateAfterReconcile = currentJobState;
@@ -119,7 +114,14 @@ public class ApplicationReconciler extends AbstractDeploymentReconciler {
if (desiredJobState == JobState.RUNNING) {
LOG.info("Upgrading/Restarting running job, suspending first...");
}
- flinkService.cancelJob(flinkApp, upgradeMode);
+ Optional<UpgradeMode> availableUpgradeMode =
+ getAvailableUpgradeMode(flinkApp, deployConfig);
+ if (availableUpgradeMode.isEmpty()) {
+ return;
+ }
+ // We must record the upgrade mode used to the status later
+ desiredJobSpec.setUpgradeMode(availableUpgradeMode.get());
+ flinkService.cancelJob(flinkApp, availableUpgradeMode.get());
stateAfterReconcile = JobState.SUSPENDED;
}
if (currentJobState == JobState.SUSPENDED && desiredJobState == JobState.RUNNING) {
@@ -128,18 +130,18 @@ public class ApplicationReconciler extends AbstractDeploymentReconciler {
desiredJobSpec,
status,
deployConfig,
- upgradeMode == UpgradeMode.LAST_STATE);
+ // We decide to enforce HA based on how job was previously suspended
+ lastReconciledSpec.getJob().getUpgradeMode() == UpgradeMode.LAST_STATE);
stateAfterReconcile = JobState.RUNNING;
}
ReconciliationUtils.updateForSpecReconciliationSuccess(flinkApp, stateAfterReconcile);
IngressUtils.updateIngressRules(
deployMeta, currentDeploySpec, deployConfig, kubernetesClient);
} else if (ReconciliationUtils.shouldRollBack(
- flinkService, reconciliationStatus, deployConfig)) {
+ flinkService, reconciliationStatus, observeConfig)) {
rollbackApplication(flinkApp);
- } else if (ReconciliationUtils.shouldRecoverDeployment(
- configManager.getObserveConfig(flinkApp), flinkApp)) {
- recoverJmDeployment(flinkApp);
+ } else if (ReconciliationUtils.shouldRecoverDeployment(observeConfig, flinkApp)) {
+ recoverJmDeployment(flinkApp, observeConfig);
} else if (SavepointUtils.shouldTriggerSavepoint(desiredJobSpec, status)
&& ReconciliationUtils.isJobRunning(status)) {
triggerSavepoint(flinkApp);
@@ -181,18 +183,20 @@ public class ApplicationReconciler extends AbstractDeploymentReconciler {
flinkApp.getMetadata(), rollbackSpec, rollbackConfig, kubernetesClient);
}
- private void recoverJmDeployment(FlinkDeployment deployment) throws Exception {
+ private void recoverJmDeployment(FlinkDeployment deployment, Configuration observeConfig)
+ throws Exception {
LOG.info("Missing Flink Cluster deployment, trying to recover...");
FlinkDeploymentSpec specToRecover = ReconciliationUtils.getDeployedSpec(deployment);
restoreJob(
deployment.getMetadata(),
specToRecover.getJob(),
deployment.getStatus(),
- configManager.getDeployConfig(deployment.getMetadata(), specToRecover),
+ observeConfig,
true);
}
- private Optional<UpgradeMode> getAvailableUpgradeMode(FlinkDeployment deployment) {
+ private Optional<UpgradeMode> getAvailableUpgradeMode(
+ FlinkDeployment deployment, Configuration deployConfig) {
var status = deployment.getStatus();
var upgradeMode = deployment.getSpec().getJob().getUpgradeMode();
var changedToLastStateWithoutHa =
@@ -220,7 +224,9 @@ public class ApplicationReconciler extends AbstractDeploymentReconciler {
}
}
- if (flinkService.isHaMetadataAvailable(configManager.getObserveConfig(deployment))) {
+ if (FlinkUtils.isKubernetesHAActivated(deployConfig)
+ && FlinkUtils.isKubernetesHAActivated(configManager.getObserveConfig(deployment))
+ && flinkService.isHaMetadataAvailable(deployConfig)) {
LOG.debug(
"Job is not running but HA metadata is available for last state restore, ready for upgrade");
return Optional.of(UpgradeMode.LAST_STATE);
@@ -245,18 +251,18 @@ public class ApplicationReconciler extends AbstractDeploymentReconciler {
ObjectMeta meta,
JobSpec jobSpec,
FlinkDeploymentStatus status,
- Configuration effectiveConfig,
+ Configuration deployConfig,
Optional<String> savepoint,
boolean requireHaMetadata)
throws Exception {
if (savepoint.isPresent()) {
- effectiveConfig.set(SavepointConfigOptions.SAVEPOINT_PATH, savepoint.get());
+ deployConfig.set(SavepointConfigOptions.SAVEPOINT_PATH, savepoint.get());
} else {
- effectiveConfig.removeConfig(SavepointConfigOptions.SAVEPOINT_PATH);
+ deployConfig.removeConfig(SavepointConfigOptions.SAVEPOINT_PATH);
}
- setRandomJobResultStorePath(effectiveConfig);
+ setRandomJobResultStorePath(deployConfig);
if (status.getJobManagerDeploymentStatus() != JobManagerDeploymentStatus.MISSING) {
if (!ReconciliationUtils.isJobInTerminalState(status)) {
@@ -267,13 +273,13 @@ public class ApplicationReconciler extends AbstractDeploymentReconciler {
flinkService.deleteClusterDeployment(meta, status, false);
FlinkUtils.waitForClusterShutdown(
kubernetesClient,
- effectiveConfig,
+ deployConfig,
configManager
.getOperatorConfiguration()
.getFlinkShutdownClusterTimeout()
.toSeconds());
}
- flinkService.submitApplicationCluster(jobSpec, effectiveConfig, requireHaMetadata);
+ flinkService.submitApplicationCluster(jobSpec, deployConfig, requireHaMetadata);
status.getJobStatus().setState(org.apache.flink.api.common.JobStatus.RECONCILING.name());
status.setJobManagerDeploymentStatus(JobManagerDeploymentStatus.DEPLOYING);
}
@@ -282,7 +288,7 @@ public class ApplicationReconciler extends AbstractDeploymentReconciler {
ObjectMeta meta,
JobSpec jobSpec,
FlinkDeploymentStatus status,
- Configuration effectiveConfig,
+ Configuration deployConfig,
boolean requireHaMetadata)
throws Exception {
Optional<String> savepointOpt = Optional.empty();
@@ -293,7 +299,7 @@ public class ApplicationReconciler extends AbstractDeploymentReconciler {
.flatMap(s -> Optional.ofNullable(s.getLocation()));
}
- deployFlinkJob(meta, jobSpec, status, effectiveConfig, savepointOpt, requireHaMetadata);
+ deployFlinkJob(meta, jobSpec, status, deployConfig, savepointOpt, requireHaMetadata);
}
// Workaround for https://issues.apache.org/jira/browse/FLINK-27569
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkService.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkService.java
index 4674ec6..124e6b5 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkService.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkService.java
@@ -140,10 +140,8 @@ public class FlinkService {
"Deploying application cluster{}",
requireHaMetadata ? " requiring last-state from HA metadata" : "");
if (FlinkUtils.isKubernetesHAActivated(conf)) {
- final String clusterId =
- Preconditions.checkNotNull(conf.get(KubernetesConfigOptions.CLUSTER_ID));
- final String namespace =
- Preconditions.checkNotNull(conf.get(KubernetesConfigOptions.NAMESPACE));
+ final String clusterId = conf.get(KubernetesConfigOptions.CLUSTER_ID);
+ final String namespace = conf.get(KubernetesConfigOptions.NAMESPACE);
// Delete the job graph in the HA ConfigMaps so that the newly changed job config(e.g.
// parallelism) could take effect
FlinkUtils.deleteJobGraphInKubernetesHA(clusterId, namespace, kubernetesClient);
@@ -169,7 +167,7 @@ public class FlinkService {
return FlinkUtils.isHaMetadataAvailable(conf, kubernetesClient);
}
- private void validateHaMetadataExists(Configuration conf) {
+ protected void validateHaMetadataExists(Configuration conf) {
if (!isHaMetadataAvailable(conf)) {
throw new DeploymentFailedException(
"HA metadata not available to restore from last state. "
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkUtils.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkUtils.java
index cc3dd72..3896f05 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkUtils.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkUtils.java
@@ -246,6 +246,7 @@ public class FlinkUtils {
public static boolean isHaMetadataAvailable(
Configuration conf, KubernetesClient kubernetesClient) {
+
String clusterId = conf.get(KubernetesConfigOptions.CLUSTER_ID);
String namespace = conf.get(KubernetesConfigOptions.NAMESPACE);
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java
index a0e20f6..2d28872 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java
@@ -38,6 +38,7 @@ import org.apache.flink.kubernetes.operator.crd.status.Savepoint;
import org.apache.flink.kubernetes.operator.exception.DeploymentFailedException;
import org.apache.flink.kubernetes.operator.observer.SavepointFetchResult;
import org.apache.flink.kubernetes.operator.service.FlinkService;
+import org.apache.flink.kubernetes.operator.utils.FlinkUtils;
import org.apache.flink.runtime.client.JobStatusMessage;
import org.apache.flink.runtime.jobgraph.SavepointConfigOptions;
import org.apache.flink.runtime.messages.Acknowledge;
@@ -115,6 +116,10 @@ public class TestingFlinkService extends FlinkService {
@Override
public void submitApplicationCluster(
JobSpec jobSpec, Configuration conf, boolean requireHaMetadata) throws Exception {
+
+ if (requireHaMetadata) {
+ validateHaMetadataExists(conf);
+ }
if (deployFailure) {
throw new DeploymentFailedException("Deployment failure", "test");
}
@@ -134,7 +139,7 @@ public class TestingFlinkService extends FlinkService {
@Override
public boolean isHaMetadataAvailable(Configuration conf) {
- return haDataAvailable;
+ return FlinkUtils.isKubernetesHAActivated(conf) && haDataAvailable;
}
public void setHaDataAvailable(boolean haDataAvailable) {
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 e0bd5d4..5c4590a 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
@@ -633,6 +633,14 @@ public class FlinkDeploymentControllerTest {
assertEquals(
JobManagerDeploymentStatus.MISSING,
appCluster.getStatus().getJobManagerDeploymentStatus());
+ assertEquals(
+ UpgradeMode.LAST_STATE,
+ appCluster
+ .getStatus()
+ .getReconciliationStatus()
+ .deserializeLastReconciledSpec()
+ .getJob()
+ .getUpgradeMode());
flinkService.setPortReady(true);
testController.reconcile(appCluster, context);