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 1f6a7505 [FLINK-29413] Make it possible to associate triggered and
completed savepoints
1f6a7505 is described below
commit 1f6a75056acae90e9fab182fd076ee6755b35bbb
Author: Matyas Orhidi <[email protected]>
AuthorDate: Tue Sep 27 13:50:06 2022 +0200
[FLINK-29413] Make it possible to associate triggered and completed
savepoints
---
docs/content/docs/custom-resource/reference.md | 14 ++
.../kubernetes/operator/crd/status/Savepoint.java | 41 ++++-
.../{Savepoint.java => SavepointFormatType.java} | 31 +---
.../operator/crd/status/SavepointInfo.java | 6 +-
.../operator/observer/SavepointObserver.java | 7 +-
.../operator/service/AbstractFlinkService.java | 37 +++--
.../kubernetes/operator/utils/SavepointUtils.java | 15 ++
.../kubernetes/operator/TestingFlinkService.java | 9 +-
.../operator/observer/SavepointObserverTest.java | 22 ++-
.../deployment/ApplicationObserverTest.java | 175 ++++++++++++++++++++-
.../sessionjob/SessionJobObserverTest.java | 6 +-
.../crds/flinkdeployments.flink.apache.org-v1.yml | 22 +++
.../crds/flinksessionjobs.flink.apache.org-v1.yml | 22 +++
13 files changed, 355 insertions(+), 52 deletions(-)
diff --git a/docs/content/docs/custom-resource/reference.md
b/docs/content/docs/custom-resource/reference.md
index 400c078a..924e89f5 100644
--- a/docs/content/docs/custom-resource/reference.md
+++ b/docs/content/docs/custom-resource/reference.md
@@ -274,6 +274,19 @@ This page serves as a full reference for FlinkDeployment
custom resource definit
| timeStamp | long | Millisecond timestamp at the start of the savepoint
operation. |
| location | java.lang.String | External pointer of the savepoint can be used
to recover jobs. |
| triggerType |
org.apache.flink.kubernetes.operator.crd.status.SavepointTriggerType |
Savepoint trigger mechanism. |
+| formatType |
org.apache.flink.kubernetes.operator.crd.status.SavepointFormatType | |
+| triggerNonce | java.lang.Long | Nonce value used when the savepoint was
triggered manually {@link SavepointTriggerType#MANUAL}, defaults to 0. |
+
+### SavepointFormatType
+**Class**: org.apache.flink.kubernetes.operator.crd.status.SavepointFormatType
+
+**Description**: Savepoint format type.
+
+| Value | Docs |
+| ----- | ---- |
+| CANONICAL | A canonical, common for all state backends format. |
+| NATIVE | A format specific for the chosen state backend. |
+| UNKNOWN | Savepoint format unknown, if the savepoint was not triggered by
the operator. |
### SavepointInfo
**Class**: org.apache.flink.kubernetes.operator.crd.status.SavepointInfo
@@ -286,6 +299,7 @@ This page serves as a full reference for FlinkDeployment
custom resource definit
| triggerId | java.lang.String | Trigger id of a pending savepoint operation. |
| triggerTimestamp | long | Trigger timestamp of a pending savepoint
operation. |
| triggerType |
org.apache.flink.kubernetes.operator.crd.status.SavepointTriggerType |
Savepoint trigger mechanism. |
+| formatType |
org.apache.flink.kubernetes.operator.crd.status.SavepointFormatType | |
| savepointHistory |
java.util.List<org.apache.flink.kubernetes.operator.crd.status.Savepoint> |
List of recent savepoints. |
| lastPeriodicSavepointTimestamp | long | Trigger timestamp of last periodic
savepoint operation. |
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/Savepoint.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/Savepoint.java
index e2681e86..6e1c51c3 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/Savepoint.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/Savepoint.java
@@ -19,14 +19,12 @@ package org.apache.flink.kubernetes.operator.crd.status;
import org.apache.flink.annotation.Experimental;
-import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
/** Represents information about a finished savepoint. */
@Experimental
@Data
-@AllArgsConstructor
@NoArgsConstructor
public class Savepoint {
/** Millisecond timestamp at the start of the savepoint operation. */
@@ -38,7 +36,42 @@ public class Savepoint {
/** Savepoint trigger mechanism. */
private SavepointTriggerType triggerType = SavepointTriggerType.UNKNOWN;
- public static Savepoint of(String location, SavepointTriggerType type) {
- return new Savepoint(System.currentTimeMillis(), location, type);
+ private SavepointFormatType formatType = SavepointFormatType.UNKNOWN;
+
+ /**
+ * Nonce value used when the savepoint was triggered manually {@link
+ * SavepointTriggerType#MANUAL}, defaults to 0.
+ */
+ private Long triggerNonce = 0L;
+
+ public Savepoint(
+ long timeStamp,
+ String location,
+ SavepointTriggerType triggerType,
+ SavepointFormatType formatType,
+ Long triggerNonce) {
+ this.timeStamp = timeStamp;
+ this.location = location;
+ this.triggerType = triggerType;
+ this.formatType = formatType;
+ setTriggerNonce(triggerNonce);
+ }
+
+ public static Savepoint of(String location, SavepointTriggerType
triggerType) {
+ return new Savepoint(
+ System.currentTimeMillis(), location, triggerType,
SavepointFormatType.UNKNOWN, 0L);
+ }
+
+ public static Savepoint of(
+ String location, SavepointTriggerType triggerType,
SavepointFormatType formatType) {
+ return new Savepoint(System.currentTimeMillis(), location,
triggerType, formatType, 0L);
+ }
+
+ public void setTriggerNonce(Long triggerNonce) {
+ if (triggerNonce == null) {
+ this.triggerNonce = 0L;
+ } else {
+ this.triggerNonce = triggerNonce;
+ }
}
}
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/Savepoint.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/SavepointFormatType.java
similarity index 53%
copy from
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/Savepoint.java
copy to
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/SavepointFormatType.java
index e2681e86..0fff963c 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/Savepoint.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/SavepointFormatType.java
@@ -17,28 +17,13 @@
package org.apache.flink.kubernetes.operator.crd.status;
-import org.apache.flink.annotation.Experimental;
+/** Savepoint format type. */
+public enum SavepointFormatType {
-import lombok.AllArgsConstructor;
-import lombok.Data;
-import lombok.NoArgsConstructor;
-
-/** Represents information about a finished savepoint. */
-@Experimental
-@Data
-@AllArgsConstructor
-@NoArgsConstructor
-public class Savepoint {
- /** Millisecond timestamp at the start of the savepoint operation. */
- private long timeStamp;
-
- /** External pointer of the savepoint can be used to recover jobs. */
- private String location;
-
- /** Savepoint trigger mechanism. */
- private SavepointTriggerType triggerType = SavepointTriggerType.UNKNOWN;
-
- public static Savepoint of(String location, SavepointTriggerType type) {
- return new Savepoint(System.currentTimeMillis(), location, type);
- }
+ /** A canonical, common for all state backends format. */
+ CANONICAL,
+ /** A format specific for the chosen state backend. */
+ NATIVE,
+ /** Savepoint format unknown, if the savepoint was not triggered by the
operator. */
+ UNKNOWN
}
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 253f8ec1..ebfd30d9 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
@@ -46,16 +46,20 @@ public class SavepointInfo {
/** Savepoint trigger mechanism. */
private SavepointTriggerType triggerType = SavepointTriggerType.UNKNOWN;
+ private SavepointFormatType formatType = SavepointFormatType.UNKNOWN;
+
/** List of recent savepoints. */
private List<Savepoint> savepointHistory = new ArrayList<>();
/** Trigger timestamp of last periodic savepoint operation. */
private long lastPeriodicSavepointTimestamp = 0L;
- public void setTrigger(String triggerId, SavepointTriggerType triggerType)
{
+ public void setTrigger(
+ String triggerId, SavepointTriggerType triggerType,
SavepointFormatType formatType) {
this.triggerId = triggerId;
this.triggerTimestamp = System.currentTimeMillis();
this.triggerType = triggerType;
+ this.formatType = formatType;
}
public void resetTrigger() {
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/SavepointObserver.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/SavepointObserver.java
index aa2d69de..f0c8ee78 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/SavepointObserver.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/SavepointObserver.java
@@ -26,6 +26,7 @@ import
org.apache.flink.kubernetes.operator.crd.AbstractFlinkResource;
import org.apache.flink.kubernetes.operator.crd.status.CommonStatus;
import org.apache.flink.kubernetes.operator.crd.status.Savepoint;
import org.apache.flink.kubernetes.operator.crd.status.SavepointInfo;
+import org.apache.flink.kubernetes.operator.crd.status.SavepointTriggerType;
import org.apache.flink.kubernetes.operator.exception.ReconciliationException;
import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
import org.apache.flink.kubernetes.operator.service.FlinkService;
@@ -127,7 +128,11 @@ public class SavepointObserver<
new Savepoint(
savepointInfo.getTriggerTimestamp(),
savepointFetchResult.getLocation(),
- savepointInfo.getTriggerType());
+ savepointInfo.getTriggerType(),
+ savepointInfo.getFormatType(),
+ SavepointTriggerType.MANUAL ==
savepointInfo.getTriggerType()
+ ?
resource.getSpec().getJob().getSavepointTriggerNonce()
+ : null);
ReconciliationUtils.updateLastReconciledSavepointTriggerNonce(savepointInfo,
resource);
savepointInfo.updateLastSavepoint(savepoint);
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java
index 0b60fabc..6ae12f80 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java
@@ -38,11 +38,13 @@ import
org.apache.flink.kubernetes.operator.crd.spec.JobSpec;
import org.apache.flink.kubernetes.operator.crd.spec.UpgradeMode;
import
org.apache.flink.kubernetes.operator.crd.status.JobManagerDeploymentStatus;
import org.apache.flink.kubernetes.operator.crd.status.Savepoint;
+import org.apache.flink.kubernetes.operator.crd.status.SavepointFormatType;
import org.apache.flink.kubernetes.operator.crd.status.SavepointTriggerType;
import
org.apache.flink.kubernetes.operator.exception.DeploymentFailedException;
import org.apache.flink.kubernetes.operator.observer.SavepointFetchResult;
import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
import org.apache.flink.kubernetes.operator.utils.FlinkUtils;
+import org.apache.flink.kubernetes.operator.utils.SavepointUtils;
import org.apache.flink.runtime.client.JobStatusMessage;
import org.apache.flink.runtime.execution.ExecutionState;
import
org.apache.flink.runtime.highavailability.nonha.standalone.StandaloneClientHAServices;
@@ -53,7 +55,6 @@ import
org.apache.flink.runtime.messages.webmonitor.MultipleJobsDetails;
import org.apache.flink.runtime.rest.FileUpload;
import org.apache.flink.runtime.rest.RestClient;
import org.apache.flink.runtime.rest.handler.async.AsynchronousOperationResult;
-import org.apache.flink.runtime.rest.handler.async.TriggerResponse;
import org.apache.flink.runtime.rest.messages.DashboardConfiguration;
import org.apache.flink.runtime.rest.messages.EmptyMessageParameters;
import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
@@ -65,7 +66,6 @@ import
org.apache.flink.runtime.rest.messages.job.savepoints.SavepointInfo;
import
org.apache.flink.runtime.rest.messages.job.savepoints.SavepointStatusHeaders;
import
org.apache.flink.runtime.rest.messages.job.savepoints.SavepointStatusMessageParameters;
import
org.apache.flink.runtime.rest.messages.job.savepoints.SavepointTriggerHeaders;
-import
org.apache.flink.runtime.rest.messages.job.savepoints.SavepointTriggerMessageParameters;
import
org.apache.flink.runtime.rest.messages.job.savepoints.SavepointTriggerRequestBody;
import org.apache.flink.runtime.rest.util.RestConstants;
import
org.apache.flink.runtime.state.memory.NonPersistentMetadataCheckpointStorageLocation;
@@ -259,6 +259,7 @@ public abstract class AbstractFlinkService implements
FlinkService {
var jobId = jobIdString != null ? JobID.fromHexString(jobIdString) :
null;
Optional<String> savepointOpt = Optional.empty();
+ var savepointFormatType = SavepointUtils.getSavepointFormatType(conf);
try (ClusterClient<String> clusterClient = getClusterClient(conf)) {
var clusterId = clusterClient.getClusterId();
switch (upgradeMode) {
@@ -300,9 +301,7 @@ public abstract class AbstractFlinkService implements
FlinkService {
conf.get(FLINK_VERSION)
.isNewerVersionThan(
FlinkVersion.v1_14)
- ? conf.get(
-
KubernetesOperatorConfigOptions
-
.OPERATOR_SAVEPOINT_FORMAT_TYPE)
+ ?
savepointFormatType
: null)
.get(timeout, TimeUnit.SECONDS);
savepointOpt = Optional.of(savepoint);
@@ -338,7 +337,11 @@ public abstract class AbstractFlinkService implements
FlinkService {
deploymentStatus.getJobStatus().setState(JobStatus.FINISHED.name());
savepointOpt.ifPresent(
location -> {
- Savepoint sp = Savepoint.of(location,
SavepointTriggerType.UPGRADE);
+ Savepoint sp =
+ Savepoint.of(
+ location,
+ SavepointTriggerType.UPGRADE,
+
SavepointFormatType.valueOf(savepointFormatType.name()));
deploymentStatus.getJobStatus().getSavepointInfo().updateLastSavepoint(sp);
});
@@ -436,16 +439,19 @@ public abstract class AbstractFlinkService implements
FlinkService {
LOG.info("Triggering new savepoint");
try (RestClusterClient<String> clusterClient =
(RestClusterClient<String>) getClusterClient(conf)) {
- SavepointTriggerHeaders savepointTriggerHeaders =
SavepointTriggerHeaders.getInstance();
- SavepointTriggerMessageParameters
savepointTriggerMessageParameters =
+ var savepointTriggerHeaders =
SavepointTriggerHeaders.getInstance();
+ var savepointTriggerMessageParameters =
savepointTriggerHeaders.getUnresolvedMessageParameters();
savepointTriggerMessageParameters.jobID.resolve(JobID.fromHexString(jobId));
- final String savepointDirectory =
+ var savepointDirectory =
Preconditions.checkNotNull(conf.get(CheckpointingOptions.SAVEPOINT_DIRECTORY));
- final long timeout =
+ var timeout =
configManager.getOperatorConfiguration().getFlinkClientTimeout().getSeconds();
- TriggerResponse response =
+
+ var savepointFormatType =
SavepointUtils.getSavepointFormatType(conf);
+
+ var response =
clusterClient
.sendRequest(
savepointTriggerHeaders,
@@ -455,15 +461,16 @@ public abstract class AbstractFlinkService implements
FlinkService {
false,
conf.get(FLINK_VERSION)
.isNewerVersionThan(FlinkVersion.v1_14)
- ? conf.get(
-
KubernetesOperatorConfigOptions
-
.OPERATOR_SAVEPOINT_FORMAT_TYPE)
+ ? savepointFormatType
: null,
null))
.get(timeout, TimeUnit.SECONDS);
LOG.info("Savepoint successfully triggered: " +
response.getTriggerId().toHexString());
- savepointInfo.setTrigger(response.getTriggerId().toHexString(),
triggerType);
+ savepointInfo.setTrigger(
+ response.getTriggerId().toHexString(),
+ triggerType,
+ SavepointFormatType.valueOf(savepointFormatType.name()));
}
}
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 2fab2efe..bc56c638 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
@@ -19,8 +19,10 @@ package org.apache.flink.kubernetes.operator.utils;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.execution.SavepointFormatType;
import
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
import org.apache.flink.kubernetes.operator.crd.AbstractFlinkResource;
+import org.apache.flink.kubernetes.operator.crd.spec.FlinkVersion;
import org.apache.flink.kubernetes.operator.crd.status.JobStatus;
import org.apache.flink.kubernetes.operator.crd.status.SavepointInfo;
import org.apache.flink.kubernetes.operator.crd.status.SavepointTriggerType;
@@ -35,6 +37,8 @@ import java.time.Duration;
import java.time.Instant;
import java.util.Optional;
+import static
org.apache.flink.kubernetes.operator.config.FlinkConfigBuilder.FLINK_VERSION;
+
/** Savepoint utilities. */
public class SavepointUtils {
@@ -164,4 +168,15 @@ public class SavepointUtils {
? "Periodic savepoint failed"
: "Savepoint failed for savepointTriggerNonce: " +
triggerNonce;
}
+
+ public static SavepointFormatType getSavepointFormatType(Configuration
configuration) {
+ var savepointFormatType =
org.apache.flink.core.execution.SavepointFormatType.CANONICAL;
+ if (configuration.get(FLINK_VERSION) != null
+ &&
configuration.get(FLINK_VERSION).isNewerVersionThan(FlinkVersion.v1_14)) {
+ savepointFormatType =
+ configuration.get(
+
KubernetesOperatorConfigOptions.OPERATOR_SAVEPOINT_FORMAT_TYPE);
+ }
+ return savepointFormatType;
+ }
}
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 27b453d0..961830c6 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
@@ -37,6 +37,7 @@ import
org.apache.flink.kubernetes.operator.crd.spec.UpgradeMode;
import org.apache.flink.kubernetes.operator.crd.status.FlinkDeploymentStatus;
import
org.apache.flink.kubernetes.operator.crd.status.JobManagerDeploymentStatus;
import org.apache.flink.kubernetes.operator.crd.status.Savepoint;
+import org.apache.flink.kubernetes.operator.crd.status.SavepointFormatType;
import org.apache.flink.kubernetes.operator.crd.status.SavepointInfo;
import org.apache.flink.kubernetes.operator.crd.status.SavepointTriggerType;
import
org.apache.flink.kubernetes.operator.exception.DeploymentFailedException;
@@ -44,6 +45,7 @@ import
org.apache.flink.kubernetes.operator.observer.SavepointFetchResult;
import org.apache.flink.kubernetes.operator.service.AbstractFlinkService;
import
org.apache.flink.kubernetes.operator.standalone.StandaloneKubernetesConfigOptionsInternal;
import org.apache.flink.kubernetes.operator.utils.FlinkUtils;
+import org.apache.flink.kubernetes.operator.utils.SavepointUtils;
import org.apache.flink.runtime.client.JobStatusMessage;
import org.apache.flink.runtime.execution.ExecutionState;
import org.apache.flink.runtime.jobgraph.SavepointConfigOptions;
@@ -127,6 +129,8 @@ public class TestingFlinkService extends
AbstractFlinkService {
public void clear() {
jobs.clear();
sessions.clear();
+ triggerCounter = 0;
+ savepointCounter = 0;
}
public Set<String> getSessions() {
@@ -253,7 +257,10 @@ public class TestingFlinkService extends
AbstractFlinkService {
SavepointInfo savepointInfo,
Configuration conf) {
var triggerId = "trigger_" + triggerCounter++;
- savepointInfo.setTrigger(triggerId, triggerType);
+
+ var savepointFormatType = SavepointUtils.getSavepointFormatType(conf);
+ savepointInfo.setTrigger(
+ triggerId, triggerType,
SavepointFormatType.valueOf(savepointFormatType.name()));
savepointTriggers.put(triggerId, false);
}
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/SavepointObserverTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/SavepointObserverTest.java
index af42289f..222fc0e6 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/SavepointObserverTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/SavepointObserverTest.java
@@ -25,6 +25,7 @@ import
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptio
import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
import org.apache.flink.kubernetes.operator.crd.status.FlinkDeploymentStatus;
import org.apache.flink.kubernetes.operator.crd.status.Savepoint;
+import org.apache.flink.kubernetes.operator.crd.status.SavepointFormatType;
import org.apache.flink.kubernetes.operator.crd.status.SavepointInfo;
import org.apache.flink.kubernetes.operator.crd.status.SavepointTriggerType;
import org.apache.flink.kubernetes.operator.utils.EventRecorder;
@@ -63,7 +64,9 @@ public class SavepointObserverTest {
SavepointInfo spInfo = new SavepointInfo();
Assertions.assertTrue(spInfo.getSavepointHistory().isEmpty());
- Savepoint sp = new Savepoint(1, "sp1", SavepointTriggerType.MANUAL);
+ Savepoint sp =
+ new Savepoint(
+ 1, "sp1", SavepointTriggerType.MANUAL,
SavepointFormatType.CANONICAL, 123L);
spInfo.updateLastSavepoint(sp);
observer.cleanupSavepointHistory(spInfo,
configManager.getDefaultConfig());
@@ -81,7 +84,9 @@ public class SavepointObserverTest {
SavepointInfo spInfo = new SavepointInfo();
- Savepoint sp1 = new Savepoint(1, "sp1", SavepointTriggerType.MANUAL);
+ Savepoint sp1 =
+ new Savepoint(
+ 1, "sp1", SavepointTriggerType.MANUAL,
SavepointFormatType.CANONICAL, 123L);
spInfo.updateLastSavepoint(sp1);
observer.cleanupSavepointHistory(spInfo, conf);
Assertions.assertIterableEquals(
@@ -89,7 +94,9 @@ public class SavepointObserverTest {
Assertions.assertIterableEquals(
Collections.emptyList(), flinkService.getDisposedSavepoints());
- Savepoint sp2 = new Savepoint(2, "sp2", SavepointTriggerType.MANUAL);
+ Savepoint sp2 =
+ new Savepoint(
+ 2, "sp2", SavepointTriggerType.MANUAL,
SavepointFormatType.CANONICAL, 123L);
spInfo.updateLastSavepoint(sp2);
observer.cleanupSavepointHistory(spInfo, conf);
Assertions.assertIterableEquals(
@@ -110,7 +117,9 @@ public class SavepointObserverTest {
configManager.updateDefaultConfig(conf);
SavepointInfo spInfo = new SavepointInfo();
- Savepoint sp1 = new Savepoint(1, "sp1", SavepointTriggerType.MANUAL);
+ Savepoint sp1 =
+ new Savepoint(
+ 1, "sp1", SavepointTriggerType.MANUAL,
SavepointFormatType.CANONICAL, 123L);
spInfo.updateLastSavepoint(sp1);
observer.cleanupSavepointHistory(spInfo, conf);
Assertions.assertIterableEquals(
@@ -118,7 +127,9 @@ public class SavepointObserverTest {
Assertions.assertIterableEquals(
Collections.emptyList(), flinkService.getDisposedSavepoints());
- Savepoint sp2 = new Savepoint(2, "sp2", SavepointTriggerType.MANUAL);
+ Savepoint sp2 =
+ new Savepoint(
+ 2, "sp2", SavepointTriggerType.MANUAL,
SavepointFormatType.CANONICAL, 123L);
spInfo.updateLastSavepoint(sp2);
observer.cleanupSavepointHistory(spInfo, conf);
Assertions.assertIterableEquals(
@@ -159,5 +170,6 @@ public class SavepointObserverTest {
assertEquals(savepointInfo.getLastSavepoint(),
savepointInfo.getSavepointHistory().get(0));
assertEquals(
SavepointTriggerType.PERIODIC,
savepointInfo.getLastSavepoint().getTriggerType());
+ assertEquals(0L, savepointInfo.getLastSavepoint().getTriggerNonce());
}
}
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserverTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserverTest.java
index c4575aa9..44570f39 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserverTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserverTest.java
@@ -24,10 +24,12 @@ import
org.apache.flink.kubernetes.operator.TestingFlinkService;
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.FlinkVersion;
import org.apache.flink.kubernetes.operator.crd.spec.JobState;
import
org.apache.flink.kubernetes.operator.crd.status.JobManagerDeploymentStatus;
import org.apache.flink.kubernetes.operator.crd.status.JobStatus;
import org.apache.flink.kubernetes.operator.crd.status.ReconciliationState;
+import org.apache.flink.kubernetes.operator.crd.status.SavepointFormatType;
import org.apache.flink.kubernetes.operator.crd.status.SavepointTriggerType;
import
org.apache.flink.kubernetes.operator.exception.DeploymentFailedException;
import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
@@ -46,8 +48,10 @@ import org.junit.jupiter.api.Test;
import java.time.Duration;
import java.time.Instant;
import java.util.HashMap;
+import java.util.Map;
import java.util.stream.Collectors;
+import static
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.OPERATOR_SAVEPOINT_FORMAT_TYPE;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
@@ -68,6 +72,7 @@ public class ApplicationObserverTest {
public void before() {
var eventRecorder = new EventRecorder(kubernetesClient, (r, e) -> {});
observer = new ApplicationObserver(flinkService, configManager,
eventRecorder);
+ flinkService.clear();
}
@Test
@@ -224,6 +229,8 @@ public class ApplicationObserverTest {
@Test
public void observeSavepoint() throws Exception {
FlinkDeployment deployment = TestUtils.buildApplicationCluster();
+ Long timedOutNonce = 1L;
+ deployment.getSpec().getJob().setSavepointTriggerNonce(timedOutNonce);
Configuration conf =
configManager.getDeployConfig(deployment.getMetadata(),
deployment.getSpec());
flinkService.submitApplicationCluster(deployment.getSpec().getJob(),
conf, false);
@@ -283,17 +290,31 @@ public class ApplicationObserverTest {
kubernetesClient.v1().events().inNamespace(deployment.getMetadata().getNamespace())
.list().getItems().stream()
.filter(e -> e.getReason().contains("SavepointError"))
+ .filter(
+ e ->
+ e.getMessage()
+ .equals(
+ "Savepoint failed for
savepointTriggerNonce: "
+ +
timedOutNonce))
.count());
assertEquals(
1,
kubernetesClient.v1().events().inNamespace(deployment.getMetadata().getNamespace())
.list().getItems().stream()
.filter(e -> e.getReason().contains("SavepointError"))
+ .filter(
+ e ->
+ e.getMessage()
+ .equals(
+ "Savepoint failed for
savepointTriggerNonce: "
+ +
timedOutNonce))
.collect(Collectors.toList())
.get(0)
.getCount());
// savepoint success
+ Long firstNonce = 123L;
+ deployment.getSpec().getJob().setSavepointTriggerNonce(firstNonce);
flinkService.triggerSavepoint(
deployment.getStatus().getJobStatus().getJobId(),
SavepointTriggerType.MANUAL,
@@ -313,9 +334,19 @@ public class ApplicationObserverTest {
.getSavepointInfo()
.getLastSavepoint()
.getLocation());
+ assertEquals(
+ firstNonce,
+ deployment
+ .getStatus()
+ .getJobStatus()
+ .getSavepointInfo()
+ .getLastSavepoint()
+ .getTriggerNonce());
assertFalse(SavepointUtils.savepointInProgress(deployment.getStatus().getJobStatus()));
// second attempt success
+ Long secondNonce = 456L;
+ deployment.getSpec().getJob().setSavepointTriggerNonce(secondNonce);
flinkService.triggerSavepoint(
deployment.getStatus().getJobStatus().getJobId(),
SavepointTriggerType.MANUAL,
@@ -337,9 +368,19 @@ public class ApplicationObserverTest {
.getSavepointInfo()
.getLastSavepoint()
.getLocation());
+ assertEquals(
+ secondNonce,
+ deployment
+ .getStatus()
+ .getJobStatus()
+ .getSavepointInfo()
+ .getLastSavepoint()
+ .getTriggerNonce());
assertFalse(SavepointUtils.savepointInProgress(deployment.getStatus().getJobStatus()));
// application failure after checkpoint trigger
+ Long thirdNonce = 789L;
+ deployment.getSpec().getJob().setSavepointTriggerNonce(thirdNonce);
flinkService.triggerSavepoint(
deployment.getStatus().getJobStatus().getJobId(),
SavepointTriggerType.MANUAL,
@@ -359,6 +400,14 @@ public class ApplicationObserverTest {
.getSavepointInfo()
.getLastSavepoint()
.getLocation());
+ assertEquals(
+ secondNonce,
+ deployment
+ .getStatus()
+ .getJobStatus()
+ .getSavepointInfo()
+ .getLastSavepoint()
+ .getTriggerNonce());
assertFalse(SavepointUtils.savepointInProgress(deployment.getStatus().getJobStatus()));
assertEquals(
@@ -366,12 +415,24 @@ public class ApplicationObserverTest {
kubernetesClient.v1().events().inNamespace(deployment.getMetadata().getNamespace())
.list().getItems().stream()
.filter(e -> e.getReason().contains("SavepointError"))
+ .filter(
+ e ->
+ e.getMessage()
+ .equals(
+ "Savepoint failed for
savepointTriggerNonce: "
+ + thirdNonce))
.count());
assertEquals(
- 2,
+ 1,
kubernetesClient.v1().events().inNamespace(deployment.getMetadata().getNamespace())
.list().getItems().stream()
.filter(e -> e.getReason().contains("SavepointError"))
+ .filter(
+ e ->
+ e.getMessage()
+ .equals(
+ "Savepoint failed for
savepointTriggerNonce: "
+ + thirdNonce))
.collect(Collectors.toList())
.get(0)
.getCount());
@@ -431,6 +492,118 @@ public class ApplicationObserverTest {
.size());
}
+ @Test
+ public void testSavepointFormat() throws Exception {
+ FlinkDeployment deployment = TestUtils.buildApplicationCluster();
+ Configuration conf =
+ configManager.getDeployConfig(deployment.getMetadata(),
deployment.getSpec());
+ flinkService.submitApplicationCluster(deployment.getSpec().getJob(),
conf, false);
+ bringToReadyStatus(deployment);
+ assertTrue(ReconciliationUtils.isJobRunning(deployment.getStatus()));
+
+ // canonical savepoint
+ Long firstNonce = 123L;
+ deployment.getSpec().getJob().setSavepointTriggerNonce(firstNonce);
+ flinkService.triggerSavepoint(
+ deployment.getStatus().getJobStatus().getJobId(),
+ SavepointTriggerType.MANUAL,
+ deployment.getStatus().getJobStatus().getSavepointInfo(),
+ conf);
+
+ observer.observe(deployment, readyContext);
+ observer.observe(deployment, readyContext);
+
assertFalse(SavepointUtils.savepointInProgress(deployment.getStatus().getJobStatus()));
+ assertEquals(
+ firstNonce,
+ deployment
+ .getStatus()
+ .getJobStatus()
+ .getSavepointInfo()
+ .getLastSavepoint()
+ .getTriggerNonce());
+ assertEquals(
+ SavepointFormatType.CANONICAL,
+ deployment
+ .getStatus()
+ .getJobStatus()
+ .getSavepointInfo()
+ .getLastSavepoint()
+ .getFormatType());
+
+ // native savepoint
+ Long secondNonce = 456L;
+ deployment.getSpec().getJob().setSavepointTriggerNonce(secondNonce);
+ deployment
+ .getSpec()
+ .setFlinkConfiguration(
+ Map.of(
+ OPERATOR_SAVEPOINT_FORMAT_TYPE.key(),
+
org.apache.flink.core.execution.SavepointFormatType.NATIVE.name()));
+ conf = configManager.getDeployConfig(deployment.getMetadata(),
deployment.getSpec());
+ flinkService.triggerSavepoint(
+ deployment.getStatus().getJobStatus().getJobId(),
+ SavepointTriggerType.MANUAL,
+ deployment.getStatus().getJobStatus().getSavepointInfo(),
+ conf);
+
+ observer.observe(deployment, readyContext);
+ observer.observe(deployment, readyContext);
+
assertFalse(SavepointUtils.savepointInProgress(deployment.getStatus().getJobStatus()));
+ assertEquals(
+ secondNonce,
+ deployment
+ .getStatus()
+ .getJobStatus()
+ .getSavepointInfo()
+ .getLastSavepoint()
+ .getTriggerNonce());
+ assertEquals(
+ SavepointFormatType.NATIVE,
+ deployment
+ .getStatus()
+ .getJobStatus()
+ .getSavepointInfo()
+ .getLastSavepoint()
+ .getFormatType());
+
+ // canonical for flink savepoint
+ Long thirdNonce = 789L;
+ deployment.getSpec().getJob().setSavepointTriggerNonce(thirdNonce);
+ deployment.getSpec().setFlinkVersion(FlinkVersion.v1_14);
+ deployment
+ .getSpec()
+ .setFlinkConfiguration(
+ Map.of(
+ OPERATOR_SAVEPOINT_FORMAT_TYPE.key(),
+
org.apache.flink.core.execution.SavepointFormatType.NATIVE.name()));
+ conf = configManager.getDeployConfig(deployment.getMetadata(),
deployment.getSpec());
+ flinkService.triggerSavepoint(
+ deployment.getStatus().getJobStatus().getJobId(),
+ SavepointTriggerType.MANUAL,
+ deployment.getStatus().getJobStatus().getSavepointInfo(),
+ conf);
+
+ observer.observe(deployment, readyContext);
+ observer.observe(deployment, readyContext);
+
assertFalse(SavepointUtils.savepointInProgress(deployment.getStatus().getJobStatus()));
+ assertEquals(
+ thirdNonce,
+ deployment
+ .getStatus()
+ .getJobStatus()
+ .getSavepointInfo()
+ .getLastSavepoint()
+ .getTriggerNonce());
+ assertEquals(
+ SavepointFormatType.CANONICAL,
+ deployment
+ .getStatus()
+ .getJobStatus()
+ .getSavepointInfo()
+ .getLastSavepoint()
+ .getFormatType());
+ }
+
private void bringToReadyStatus(FlinkDeployment deployment) {
ReconciliationUtils.updateStatusForDeployedSpec(deployment, new
Configuration());
JobStatus jobStatus = new JobStatus();
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/sessionjob/SessionJobObserverTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/sessionjob/SessionJobObserverTest.java
index 17075ed6..f42287f4 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/sessionjob/SessionJobObserverTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/sessionjob/SessionJobObserverTest.java
@@ -197,7 +197,6 @@ public class SessionJobObserverTest {
public void testObserveSavepoint() throws Exception {
final var sessionJob = TestUtils.buildSessionJob();
final var readyContext =
TestUtils.createContextWithReadyFlinkDeployment();
-
// submit job
reconciler.reconcile(sessionJob, readyContext);
var jobID = sessionJob.getStatus().getJobStatus().getJobId();
@@ -213,12 +212,16 @@ public class SessionJobObserverTest {
Assertions.assertFalse(
SavepointUtils.savepointInProgress(sessionJob.getStatus().getJobStatus()));
+ Long firstNonce = 123L;
+ sessionJob.getSpec().getJob().setSavepointTriggerNonce(firstNonce);
flinkService.triggerSavepoint(
jobID, SavepointTriggerType.MANUAL, savepointInfo, new
Configuration());
Assertions.assertTrue(
SavepointUtils.savepointInProgress(sessionJob.getStatus().getJobStatus()));
Assertions.assertEquals("trigger_0", savepointInfo.getTriggerId());
+ Long secondNonce = 456L;
+ sessionJob.getSpec().getJob().setSavepointTriggerNonce(secondNonce);
flinkService.triggerSavepoint(
jobID, SavepointTriggerType.MANUAL, savepointInfo, new
Configuration());
Assertions.assertTrue(
@@ -231,6 +234,7 @@ public class SessionJobObserverTest {
observer.observe(sessionJob, readyContext); // pending
observer.observe(sessionJob, readyContext); // success
Assertions.assertEquals("savepoint_0",
savepointInfo.getLastSavepoint().getLocation());
+ Assertions.assertEquals(secondNonce,
savepointInfo.getLastSavepoint().getTriggerNonce());
Assertions.assertFalse(
SavepointUtils.savepointInProgress(sessionJob.getStatus().getJobStatus()));
}
diff --git
a/helm/flink-kubernetes-operator/crds/flinkdeployments.flink.apache.org-v1.yml
b/helm/flink-kubernetes-operator/crds/flinkdeployments.flink.apache.org-v1.yml
index ba33f580..4f0563db 100644
---
a/helm/flink-kubernetes-operator/crds/flinkdeployments.flink.apache.org-v1.yml
+++
b/helm/flink-kubernetes-operator/crds/flinkdeployments.flink.apache.org-v1.yml
@@ -9162,6 +9162,14 @@ spec:
- UPGRADE
- UNKNOWN
type: string
+ formatType:
+ enum:
+ - CANONICAL
+ - NATIVE
+ - UNKNOWN
+ type: string
+ triggerNonce:
+ type: integer
type: object
triggerId:
type: string
@@ -9174,6 +9182,12 @@ spec:
- UPGRADE
- UNKNOWN
type: string
+ formatType:
+ enum:
+ - CANONICAL
+ - NATIVE
+ - UNKNOWN
+ type: string
savepointHistory:
items:
properties:
@@ -9188,6 +9202,14 @@ spec:
- UPGRADE
- UNKNOWN
type: string
+ formatType:
+ enum:
+ - CANONICAL
+ - NATIVE
+ - UNKNOWN
+ type: string
+ triggerNonce:
+ type: integer
type: object
type: array
lastPeriodicSavepointTimestamp:
diff --git
a/helm/flink-kubernetes-operator/crds/flinksessionjobs.flink.apache.org-v1.yml
b/helm/flink-kubernetes-operator/crds/flinksessionjobs.flink.apache.org-v1.yml
index ec20e56c..1ab22032 100644
---
a/helm/flink-kubernetes-operator/crds/flinksessionjobs.flink.apache.org-v1.yml
+++
b/helm/flink-kubernetes-operator/crds/flinksessionjobs.flink.apache.org-v1.yml
@@ -112,6 +112,14 @@ spec:
- UPGRADE
- UNKNOWN
type: string
+ formatType:
+ enum:
+ - CANONICAL
+ - NATIVE
+ - UNKNOWN
+ type: string
+ triggerNonce:
+ type: integer
type: object
triggerId:
type: string
@@ -124,6 +132,12 @@ spec:
- UPGRADE
- UNKNOWN
type: string
+ formatType:
+ enum:
+ - CANONICAL
+ - NATIVE
+ - UNKNOWN
+ type: string
savepointHistory:
items:
properties:
@@ -138,6 +152,14 @@ spec:
- UPGRADE
- UNKNOWN
type: string
+ formatType:
+ enum:
+ - CANONICAL
+ - NATIVE
+ - UNKNOWN
+ type: string
+ triggerNonce:
+ type: integer
type: object
type: array
lastPeriodicSavepointTimestamp:
