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/04/24 11:32:05 UTC
[flink-kubernetes-operator] branch main updated: [FLINK-26926] Allow users to force upgrade even if savepoint is in progress
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 d1b20f1 [FLINK-26926] Allow users to force upgrade even if savepoint is in progress
d1b20f1 is described below
commit d1b20f1a500c75b54b34a86c6d09bf92e300a9fd
Author: Nicholas Jiang <pr...@163.com>
AuthorDate: Sun Apr 24 19:32:00 2022 +0800
[FLINK-26926] Allow users to force upgrade even if savepoint is in progress
---
docs/content/docs/operations/configuration.md | 3 ++
.../config/KubernetesOperatorConfigOptions.java | 6 +++
.../deployment/ApplicationReconciler.java | 7 +++-
.../sessionjob/FlinkSessionJobReconciler.java | 10 +++--
.../deployment/ApplicationReconcilerTest.java | 38 +++++++++++++++++
.../sessionjob/FlinkSessionJobReconcilerTest.java | 48 ++++++++++++++++++++++
.../flink-kubernetes-operator/conf/flink-conf.yaml | 4 ++
7 files changed, 110 insertions(+), 6 deletions(-)
diff --git a/docs/content/docs/operations/configuration.md b/docs/content/docs/operations/configuration.md
index 5a4e54d..6b31350 100644
--- a/docs/content/docs/operations/configuration.md
+++ b/docs/content/docs/operations/configuration.md
@@ -62,4 +62,7 @@ To learn more about metrics and logging configuration please refer to the dedica
| kubernetes.operator.observer.flink.client.timeout | 10s | Duration | The timeout for the observer to wait the flink rest client to return. |
| kubernetes.operator.reconciler.flink.cancel.job.timeout | 1min | Duration | The timeout for the reconciler to wait for flink to cancel job. |
| kubernetes.operator.reconciler.flink.cluster.shutdown.timeout | 60s | Duration | The timeout for the reconciler to wait for flink to shutdown cluster. |
+| kubernetes.operator.deployment.rollback.enabled | false | Boolean | Whether to enable rolling back failed deployment upgrades. |
+| kubernetes.operator.deployment.readiness.timeout | 1min | Duration | The timeout for deployments to become ready/stable before being rolled back if rollback is enabled. |
| kubernetes.operator.user.artifacts.base.dir | /opt/flink/artifacts | String | The base dir to put the session job artifacts. |
+| kubernetes.operator.job.upgrade.ignore-pending-savepoint | false | Boolean | Whether to ignore pending savepoint during job upgrade. |
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 aeace06..0bc7a2c 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
@@ -103,4 +103,10 @@ public class KubernetesOperatorConfigOptions {
.stringType()
.defaultValue("/opt/flink/artifacts")
.withDescription("The base dir to put the session job artifacts.");
+
+ public static final ConfigOption<Boolean> JOB_UPGRADE_IGNORE_PENDING_SAVEPOINT =
+ ConfigOptions.key("kubernetes.operator.job.upgrade.ignore-pending-savepoint")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Whether to ignore pending savepoint during job upgrade.");
}
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 ed82d37..a1bb6e0 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
@@ -20,6 +20,7 @@ package org.apache.flink.kubernetes.operator.reconciler.deployment;
import org.apache.flink.api.common.JobID;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.kubernetes.operator.config.FlinkOperatorConfiguration;
+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.JobSpec;
@@ -89,8 +90,10 @@ public class ApplicationReconciler extends AbstractDeploymentReconciler {
return;
}
- if (SavepointUtils.savepointInProgress(status.getJobStatus())) {
- LOG.info("Delaying job reconciliation until pending savepoint is completed");
+ if (!effectiveConfig.getBoolean(
+ KubernetesOperatorConfigOptions.JOB_UPGRADE_IGNORE_PENDING_SAVEPOINT)
+ && SavepointUtils.savepointInProgress(status.getJobStatus())) {
+ LOG.info("Delaying job reconciliation until pending savepoint is completed.");
return;
}
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/FlinkSessionJobReconciler.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/FlinkSessionJobReconciler.java
index 24da1b4..d60a9b7 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/FlinkSessionJobReconciler.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/FlinkSessionJobReconciler.java
@@ -20,6 +20,7 @@ package org.apache.flink.kubernetes.operator.reconciler.sessionjob;
import org.apache.flink.api.common.JobID;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.kubernetes.operator.config.FlinkOperatorConfiguration;
+import org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
import org.apache.flink.kubernetes.operator.crd.FlinkSessionJob;
import org.apache.flink.kubernetes.operator.crd.spec.FlinkSessionJobSpec;
@@ -85,9 +86,9 @@ public class FlinkSessionJobReconciler implements Reconciler<FlinkSessionJob> {
return;
}
+ Configuration effectiveConfig =
+ FlinkUtils.getEffectiveConfig(flinkDepOptional.get(), defaultConfig);
if (lastReconciledSpec == null) {
- Configuration effectiveConfig =
- FlinkUtils.getEffectiveConfig(flinkDepOptional.get(), defaultConfig);
submitAndInitStatus(
flinkSessionJob,
effectiveConfig,
@@ -99,14 +100,15 @@ public class FlinkSessionJobReconciler implements Reconciler<FlinkSessionJob> {
return;
}
- if (helper.savepointInProgress()) {
+ if (!effectiveConfig.getBoolean(
+ KubernetesOperatorConfigOptions.JOB_UPGRADE_IGNORE_PENDING_SAVEPOINT)
+ && helper.savepointInProgress()) {
LOG.info("Delaying job reconciliation until pending savepoint is completed");
return;
}
boolean specChanged = helper.specChanged(lastReconciledSpec);
- var effectiveConfig = FlinkUtils.getEffectiveConfig(flinkDepOptional.get(), defaultConfig);
if (specChanged) {
var jobSpec = flinkSessionJob.getSpec().getJob();
JobState currentJobState = lastReconciledSpec.getJob().getState();
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java
index 9ad2158..e3595cb 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerTest.java
@@ -23,6 +23,7 @@ import org.apache.flink.configuration.HighAvailabilityOptions;
import org.apache.flink.kubernetes.operator.TestUtils;
import org.apache.flink.kubernetes.operator.TestingFlinkService;
import org.apache.flink.kubernetes.operator.config.FlinkOperatorConfiguration;
+import org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
import org.apache.flink.kubernetes.operator.crd.spec.JobState;
import org.apache.flink.kubernetes.operator.crd.spec.UpgradeMode;
@@ -361,4 +362,41 @@ public class ApplicationReconcilerTest {
.build());
deployment.getStatus().setJobManagerDeploymentStatus(JobManagerDeploymentStatus.READY);
}
+
+ @Test
+ public void testJobUpgradeIgnorePendingSavepoint() throws Exception {
+ Context context = TestUtils.createContextWithReadyJobManagerDeployment();
+ TestingFlinkService flinkService = new TestingFlinkService();
+ ApplicationReconciler reconciler =
+ new ApplicationReconciler(
+ null, flinkService, operatorConfiguration, new Configuration());
+ FlinkDeployment deployment = TestUtils.buildApplicationCluster();
+ reconciler.reconcile(deployment, context);
+ List<Tuple2<String, JobStatusMessage>> runningJobs = flinkService.listJobs();
+ verifyAndSetRunningJobsToStatus(deployment, runningJobs);
+
+ FlinkDeployment spDeployment = ReconciliationUtils.clone(deployment);
+ spDeployment
+ .getSpec()
+ .getJob()
+ .setSavepointTriggerNonce(ThreadLocalRandom.current().nextLong());
+ reconciler.reconcile(spDeployment, context);
+ assertEquals(
+ "trigger_0",
+ spDeployment.getStatus().getJobStatus().getSavepointInfo().getTriggerId());
+ assertEquals(JobState.RUNNING.name(), spDeployment.getStatus().getJobStatus().getState());
+
+ // Force upgrade when savepoint is in progress.
+ spDeployment
+ .getSpec()
+ .getFlinkConfiguration()
+ .put(
+ KubernetesOperatorConfigOptions.JOB_UPGRADE_IGNORE_PENDING_SAVEPOINT.key(),
+ "true");
+ reconciler.reconcile(spDeployment, context);
+ assertEquals(
+ "trigger_0",
+ spDeployment.getStatus().getJobStatus().getSavepointInfo().getTriggerId());
+ assertEquals(JobState.SUSPENDED.name(), spDeployment.getStatus().getJobStatus().getState());
+ }
}
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/FlinkSessionJobReconcilerTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/FlinkSessionJobReconcilerTest.java
index bb87c62..9b88b2d 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/FlinkSessionJobReconcilerTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/FlinkSessionJobReconcilerTest.java
@@ -22,18 +22,21 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.kubernetes.operator.TestUtils;
import org.apache.flink.kubernetes.operator.TestingFlinkService;
import org.apache.flink.kubernetes.operator.config.FlinkOperatorConfiguration;
+import org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
import org.apache.flink.kubernetes.operator.crd.FlinkSessionJob;
import org.apache.flink.kubernetes.operator.crd.spec.JobState;
import org.apache.flink.kubernetes.operator.crd.spec.UpgradeMode;
import org.apache.flink.kubernetes.operator.crd.status.JobStatus;
import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import javax.annotation.Nullable;
import java.util.Map;
+import java.util.concurrent.ThreadLocalRandom;
import static org.apache.flink.kubernetes.operator.observer.deployment.AbstractDeploymentObserver.JOB_STATE_UNKNOWN;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -356,4 +359,49 @@ public class FlinkSessionJobReconcilerTest {
assertEquals(jobStatusObserved, sessionJob.getStatus().getJobStatus().getState());
}
+
+ @Test
+ public void testJobUpgradeIgnorePendingSavepoint() throws Exception {
+ Context readyContext = TestUtils.createContextWithReadyFlinkDeployment();
+ TestingFlinkService flinkService = new TestingFlinkService();
+ FlinkSessionJobReconciler reconciler =
+ new FlinkSessionJobReconciler(
+ null, flinkService, operatorConfiguration, defaultConfig);
+ FlinkSessionJob sessionJob = TestUtils.buildSessionJob();
+ reconciler.reconcile(sessionJob, readyContext);
+ verifyAndSetRunningJobsToStatus(
+ sessionJob,
+ JobState.RUNNING,
+ JOB_STATE_UNKNOWN,
+ null,
+ flinkService.listSessionJobs());
+
+ FlinkSessionJob spSessionJob = ReconciliationUtils.clone(sessionJob);
+ spSessionJob
+ .getSpec()
+ .getJob()
+ .setSavepointTriggerNonce(ThreadLocalRandom.current().nextLong());
+ reconciler.reconcile(spSessionJob, readyContext);
+ assertEquals(
+ "trigger_0",
+ spSessionJob.getStatus().getJobStatus().getSavepointInfo().getTriggerId());
+ assertEquals(JobState.RUNNING.name(), spSessionJob.getStatus().getJobStatus().getState());
+
+ // Force upgrade when savepoint is in progress.
+ reconciler =
+ new FlinkSessionJobReconciler(
+ null,
+ flinkService,
+ operatorConfiguration,
+ defaultConfig.set(
+ KubernetesOperatorConfigOptions
+ .JOB_UPGRADE_IGNORE_PENDING_SAVEPOINT,
+ true));
+ spSessionJob.getSpec().getJob().setParallelism(100);
+ reconciler.reconcile(spSessionJob, readyContext);
+ assertEquals(
+ "trigger_0",
+ spSessionJob.getStatus().getJobStatus().getSavepointInfo().getTriggerId());
+ assertEquals(JobState.SUSPENDED.name(), spSessionJob.getStatus().getJobStatus().getState());
+ }
}
diff --git a/helm/flink-kubernetes-operator/conf/flink-conf.yaml b/helm/flink-kubernetes-operator/conf/flink-conf.yaml
index fcbea15..8aa5835 100644
--- a/helm/flink-kubernetes-operator/conf/flink-conf.yaml
+++ b/helm/flink-kubernetes-operator/conf/flink-conf.yaml
@@ -35,6 +35,10 @@ parallelism.default: 2
# kubernetes.operator.observer.progress-check.interval: 10 s
# kubernetes.operator.observer.savepoint.trigger.grace-period: 10 s
# kubernetes.operator.observer.flink.client.timeout: 10 s
+# kubernetes.operator.deployment.rollback.enabled: false
+# kubernetes.operator.deployment.readiness.timeout: 1min
+# kubernetes.operator.user.artifacts.base.dir: /opt/flink/artifacts
+# kubernetes.operator.job.upgrade.ignore-pending-savepoint: false
# kubernetes.operator.metrics.reporter.slf4j.factory.class: org.apache.flink.metrics.slf4j.Slf4jReporterFactory
# kubernetes.operator.metrics.reporter.slf4j.interval: 5 MINUTE