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/03/11 14:57:25 UTC
[flink-kubernetes-operator] branch main updated: [FLINK-26571] Savepoint trigger/tracking improvements
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 33572a0 [FLINK-26571] Savepoint trigger/tracking improvements
33572a0 is described below
commit 33572a0052f6c046ac06f3fb7abea864cfb20c49
Author: Matyas Orhidi <ma...@apple.com>
AuthorDate: Thu Mar 10 10:13:44 2022 +0100
[FLINK-26571] Savepoint trigger/tracking improvements
---
.../kubernetes/operator/crd/spec/JobSpec.java | 2 +-
.../operator/crd/status/SavepointInfo.java | 17 ++++++++-
.../kubernetes/operator/observer/Observer.java | 6 ++--
.../operator/reconciler/ReconciliationUtils.java | 2 +-
.../kubernetes/operator/service/FlinkService.java | 3 +-
.../kubernetes/operator/utils/SavepointUtils.java | 17 +++++----
.../kubernetes/operator/TestingFlinkService.java | 4 +--
.../kubernetes/operator/observer/ObserverTest.java | 4 ++-
.../operator/reconciler/JobReconcilerTest.java | 41 ++++++++++++++++++++--
9 files changed, 76 insertions(+), 20 deletions(-)
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/JobSpec.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/JobSpec.java
index 9a687ac..3e8bedc 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/JobSpec.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/JobSpec.java
@@ -36,7 +36,7 @@ public class JobSpec {
private JobState state = JobState.RUNNING;
// The below fields are excluded from equals to avoid triggering job upgrades on changing these
- @EqualsAndHashCode.Exclude private long savepointTriggerNonce;
+ @EqualsAndHashCode.Exclude private Long savepointTriggerNonce;
@EqualsAndHashCode.Exclude private String initialSavepointPath;
@EqualsAndHashCode.Exclude private UpgradeMode upgradeMode = UpgradeMode.STATELESS;
}
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/SavepointInfo.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/SavepointInfo.java
index 23a892e..97f1e9a 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/SavepointInfo.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/SavepointInfo.java
@@ -30,5 +30,20 @@ import lombok.NoArgsConstructor;
public class SavepointInfo {
private Savepoint lastSavepoint;
private String triggerId;
- private long triggerTimestamp;
+ private Long triggerTimestamp;
+
+ public void setTrigger(String triggerId) {
+ this.triggerId = triggerId;
+ this.triggerTimestamp = System.currentTimeMillis();
+ }
+
+ public void resetTrigger() {
+ this.triggerId = null;
+ this.triggerTimestamp = null;
+ }
+
+ public void updateLastSavepoint(Savepoint savepoint) {
+ lastSavepoint = savepoint;
+ resetTrigger();
+ }
}
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/Observer.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/Observer.java
index fc0b894..51ccf75 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/Observer.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/Observer.java
@@ -177,7 +177,7 @@ public class Observer {
|| SavepointUtils.gracePeriodEnded(operatorConfiguration, savepointInfo)) {
String errorMsg = error != null ? error : "Savepoint status unknown";
LOG.error(errorMsg);
- savepointInfo.setTriggerId(null);
+ savepointInfo.resetTrigger();
ReconciliationUtils.updateForReconciliationError(flinkApp, errorMsg);
return;
}
@@ -187,9 +187,7 @@ public class Observer {
LOG.info("Savepoint not completed yet");
return;
}
-
- savepointInfo.setLastSavepoint(savepointFetchResult.getSavepoint());
- savepointInfo.setTriggerId(null);
+ savepointInfo.updateLastSavepoint(savepointFetchResult.getSavepoint());
}
/** Update previous job status based on the job list from the cluster. */
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconciliationUtils.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconciliationUtils.java
index aa58336..8937c73 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconciliationUtils.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconciliationUtils.java
@@ -36,7 +36,7 @@ public class ReconciliationUtils {
FlinkDeploymentSpec clonedSpec = clone(flinkApp.getSpec());
if (reconciliationStatus.getLastReconciledSpec() != null
&& reconciliationStatus.getLastReconciledSpec().getJob() != null) {
- long oldSavepointTriggerNonce =
+ Long oldSavepointTriggerNonce =
reconciliationStatus
.getLastReconciledSpec()
.getJob()
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 140baea..f3e8655 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
@@ -203,8 +203,7 @@ public class FlinkService {
org.apache.flink.kubernetes.operator.crd.status.SavepointInfo savepointInfo =
deployment.getStatus().getJobStatus().getSavepointInfo();
-
- savepointInfo.setTriggerId(response.getTriggerId().toHexString());
+ savepointInfo.setTrigger(response.getTriggerId().toHexString());
savepointInfo.setTriggerTimestamp(System.currentTimeMillis());
}
}
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/SavepointUtils.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/SavepointUtils.java
index 9623c08..70390f3 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/SavepointUtils.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/SavepointUtils.java
@@ -34,13 +34,18 @@ public class SavepointUtils {
if (savepointInProgress(flinkDeployment)) {
return false;
}
- return flinkDeployment.getSpec().getJob().getSavepointTriggerNonce()
- != flinkDeployment
- .getStatus()
- .getReconciliationStatus()
- .getLastReconciledSpec()
+ return flinkDeployment.getSpec().getJob().getSavepointTriggerNonce() != null
+ && !flinkDeployment
+ .getSpec()
.getJob()
- .getSavepointTriggerNonce();
+ .getSavepointTriggerNonce()
+ .equals(
+ flinkDeployment
+ .getStatus()
+ .getReconciliationStatus()
+ .getLastReconciledSpec()
+ .getJob()
+ .getSavepointTriggerNonce());
}
public static boolean gracePeriodEnded(
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 69a7f4c..cbb7fb9 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
@@ -41,6 +41,7 @@ import java.util.stream.Collectors;
public class TestingFlinkService extends FlinkService {
private int savepointCounter = 0;
+ private int triggerCounter = 0;
private List<Tuple2<String, JobStatusMessage>> jobs = new ArrayList<>();
private Set<String> sessions = new HashSet<>();
@@ -114,8 +115,7 @@ public class TestingFlinkService extends FlinkService {
@Override
public void triggerSavepoint(FlinkDeployment deployment, Configuration conf) throws Exception {
SavepointInfo savepointInfo = deployment.getStatus().getJobStatus().getSavepointInfo();
- savepointInfo.setTriggerId("trigger_" + savepointCounter);
- savepointInfo.setTriggerTimestamp(System.currentTimeMillis());
+ savepointInfo.setTrigger("trigger_" + triggerCounter++);
}
@Override
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/ObserverTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/ObserverTest.java
index 4a3a425..ea15298 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/ObserverTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/ObserverTest.java
@@ -154,8 +154,8 @@ public class ObserverTest {
assertEquals(
JobManagerDeploymentStatus.READY,
deployment.getStatus().getJobManagerDeploymentStatus());
- flinkService.triggerSavepoint(deployment, conf);
+ flinkService.triggerSavepoint(deployment, conf);
assertEquals(
"trigger_0",
deployment.getStatus().getJobStatus().getSavepointInfo().getTriggerId());
@@ -169,6 +169,7 @@ public class ObserverTest {
.getLastSavepoint()
.getLocation());
assertNull(deployment.getStatus().getJobStatus().getSavepointInfo().getTriggerId());
+ assertNull(deployment.getStatus().getJobStatus().getSavepointInfo().getTriggerTimestamp());
flinkService.triggerSavepoint(deployment, conf);
assertEquals(
@@ -184,6 +185,7 @@ public class ObserverTest {
.getLastSavepoint()
.getLocation());
assertNull(deployment.getStatus().getJobStatus().getSavepointInfo().getTriggerId());
+ assertNull(deployment.getStatus().getJobStatus().getSavepointInfo().getTriggerTimestamp());
}
private void bringToReadyStatus(FlinkDeployment deployment) {
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/JobReconcilerTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/JobReconcilerTest.java
index 40aaf3b..2e92015 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/JobReconcilerTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/JobReconcilerTest.java
@@ -35,6 +35,7 @@ import io.javaoperatorsdk.operator.api.reconciler.Context;
import org.junit.jupiter.api.Test;
import java.util.List;
+import java.util.concurrent.ThreadLocalRandom;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
@@ -154,12 +155,48 @@ public class JobReconcilerTest {
// trigger savepoint
FlinkDeployment spDeployment = ReconciliationUtils.clone(deployment);
- long oldValue = spDeployment.getSpec().getJob().getSavepointTriggerNonce();
- spDeployment.getSpec().getJob().setSavepointTriggerNonce(oldValue + 1);
+
+ // don't trigger if nonce is missing
+ reconciler.reconcile("test", spDeployment, context, config);
+ assertNull(spDeployment.getStatus().getJobStatus().getSavepointInfo().getTriggerId());
+
+ // trigger when nonce is defined
+ spDeployment
+ .getSpec()
+ .getJob()
+ .setSavepointTriggerNonce(ThreadLocalRandom.current().nextLong());
+ reconciler.reconcile("test", spDeployment, context, config);
+ assertEquals(
+ "trigger_0",
+ spDeployment.getStatus().getJobStatus().getSavepointInfo().getTriggerId());
+
+ // don't trigger when savepoint is in progress
reconciler.reconcile("test", spDeployment, context, config);
assertEquals(
"trigger_0",
spDeployment.getStatus().getJobStatus().getSavepointInfo().getTriggerId());
+ spDeployment.getStatus().getJobStatus().getSavepointInfo().setTriggerId(null);
+
+ // don't trigger when nonce is the same
+ reconciler.reconcile("test", spDeployment, context, config);
+ assertNull(spDeployment.getStatus().getJobStatus().getSavepointInfo().getTriggerId());
+ spDeployment.getStatus().getJobStatus().getSavepointInfo().setTriggerId(null);
+
+ // trigger when new nonce is defined
+ spDeployment
+ .getSpec()
+ .getJob()
+ .setSavepointTriggerNonce(ThreadLocalRandom.current().nextLong());
+ reconciler.reconcile("test", spDeployment, context, config);
+ assertEquals(
+ "trigger_1",
+ spDeployment.getStatus().getJobStatus().getSavepointInfo().getTriggerId());
+ spDeployment.getStatus().getJobStatus().getSavepointInfo().setTriggerId(null);
+
+ // don't trigger nonce is cleared
+ spDeployment.getSpec().getJob().setSavepointTriggerNonce(null);
+ reconciler.reconcile("test", spDeployment, context, config);
+ assertNull(spDeployment.getStatus().getJobStatus().getSavepointInfo().getTriggerId());
}
private void verifyAndSetRunningJobsToStatus(