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 203367b [FLINK-26179] Support for periodic savepoints
203367b is described below
commit 203367bc5f5fe3e0b9398ccb4d3430aa4e94ac88
Author: Gyula Fora <[email protected]>
AuthorDate: Thu May 26 15:00:10 2022 -0700
[FLINK-26179] Support for periodic savepoints
---
docs/content/docs/custom-resource/reference.md | 17 ++-
.../kubernetes_operator_config_configuration.html | 6 +
.../config/KubernetesOperatorConfigOptions.java | 8 ++
.../kubernetes/operator/crd/status/Savepoint.java | 7 +-
.../operator/crd/status/SavepointInfo.java | 39 +++---
.../{Savepoint.java => SavepointTriggerType.java} | 31 ++---
.../operator/observer/JobStatusObserver.java | 2 +-
.../operator/observer/SavepointFetchResult.java | 8 +-
.../operator/observer/SavepointObserver.java | 27 ++--
.../deployment/AbstractDeploymentObserver.java | 23 +---
.../observer/sessionjob/SessionJobObserver.java | 5 +
.../operator/reconciler/ReconciliationUtils.java | 12 +-
.../deployment/ApplicationReconciler.java | 15 +--
.../sessionjob/FlinkSessionJobReconciler.java | 39 +++---
.../reconciler/sessionjob/SessionJobHelper.java | 34 -----
.../kubernetes/operator/service/FlinkService.java | 36 ++++--
.../kubernetes/operator/utils/SavepointUtils.java | 141 +++++++++++++++++++--
.../operator/validation/DefaultValidator.java | 6 +
.../flink/kubernetes/operator/TestUtils.java | 3 +
.../kubernetes/operator/TestingFlinkService.java | 48 +++----
.../operator/controller/RollbackTest.java | 4 +-
.../operator/observer/SavepointObserverTest.java | 50 +++++++-
.../deployment/ApplicationObserverTest.java | 10 ++
.../sessionjob/SessionJobObserverTest.java | 13 +-
.../deployment/ApplicationReconcilerTest.java | 17 ++-
.../sessionjob/FlinkSessionJobReconcilerTest.java | 12 ++
.../operator/service/FlinkServiceTest.java | 2 +
.../operator/utils/SavepointUtilsTest.java | 84 ++++++++++++
.../operator/validation/DefaultValidatorTest.java | 17 ++-
.../crds/flinkdeployments.flink.apache.org-v1.yml | 23 ++++
.../crds/flinksessionjobs.flink.apache.org-v1.yml | 23 ++++
31 files changed, 545 insertions(+), 217 deletions(-)
diff --git a/docs/content/docs/custom-resource/reference.md
b/docs/content/docs/custom-resource/reference.md
index a416cd6..7f44e83 100644
--- a/docs/content/docs/custom-resource/reference.md
+++ b/docs/content/docs/custom-resource/reference.md
@@ -260,6 +260,7 @@ 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. |
### SavepointInfo
**Class**: org.apache.flink.kubernetes.operator.crd.status.SavepointInfo
@@ -270,5 +271,19 @@ This page serves as a full reference for FlinkDeployment
custom resource definit
| ----------| ---- | ---- |
| lastSavepoint | org.apache.flink.kubernetes.operator.crd.status.Savepoint |
Last completed savepoint by the operator. |
| triggerId | java.lang.String | Trigger id of a pending savepoint operation. |
-| triggerTimestamp | java.lang.Long | Trigger timestamp 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. |
| 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. |
+
+### SavepointTriggerType
+**Class**: org.apache.flink.kubernetes.operator.crd.status.SavepointTriggerType
+
+**Description**: Savepoint trigger mechanism.
+
+| Value | Docs |
+| ----- | ---- |
+| MANUAL | Savepoint manually triggered by changing the savepointTriggerNonce.
|
+| PERIODIC | Savepoint periodically triggered by the operator. |
+| UPGRADE | Savepoint triggered during stateful upgrade. |
+| UNKNOWN | Savepoint trigger mechanism unknown, such as savepoint retrieved
directly from Flink job. |
diff --git
a/docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html
b/docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html
index 2e8e214..5e9b466 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>Duration</td>
<td>The interval before a savepoint trigger attempt is marked as
unsuccessful.</td>
</tr>
+ <tr>
+ <td><h5>kubernetes.operator.periodic.savepoint.interval</h5></td>
+ <td style="word-wrap: break-word;">0 ms</td>
+ <td>Duration</td>
+ <td>Interval at which periodic savepoints will be triggered. The
triggering schedule is not guaranteed, savepoints will be triggered as part of
the regular reconcile loop.</td>
+ </tr>
<tr>
<td><h5>kubernetes.operator.reconciler.flink.cancel.job.timeout</h5></td>
<td style="word-wrap: break-word;">1 min</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 797f996..f5eaade 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
@@ -163,4 +163,12 @@ public class KubernetesOperatorConfigOptions {
.withDescription(
"Custom HTTP header for HttpArtifactFetcher. The
header will be applied when getting the session job artifacts. "
+ "Expected format:
headerKey1:headerValue1,headerKey2:headerValue2.");
+
+ public static final ConfigOption<Duration> PERIODIC_SAVEPOINT_INTERVAL =
+
ConfigOptions.key("kubernetes.operator.periodic.savepoint.interval")
+ .durationType()
+ .defaultValue(Duration.ZERO)
+ .withDescription(
+ "Interval at which periodic savepoints will be
triggered. "
+ + "The triggering schedule is not
guaranteed, savepoints will be triggered as part of the regular reconcile
loop.");
}
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 73da6da..e2681e8 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
@@ -35,7 +35,10 @@ public class Savepoint {
/** External pointer of the savepoint can be used to recover jobs. */
private String location;
- public static Savepoint of(String location) {
- return new Savepoint(System.currentTimeMillis(), location);
+ /** Savepoint trigger mechanism. */
+ private SavepointTriggerType triggerType = SavepointTriggerType.UNKNOWN;
+
+ public static Savepoint of(String location, SavepointTriggerType type) {
+ return new Savepoint(System.currentTimeMillis(), location, type);
}
}
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 a2eb133..eb209fc 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
@@ -23,7 +23,6 @@ import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
-import org.apache.commons.lang3.StringUtils;
import java.util.ArrayList;
import java.util.List;
@@ -42,40 +41,42 @@ public class SavepointInfo {
private String triggerId = "";
/** Trigger timestamp of a pending savepoint operation. */
- private Long triggerTimestamp = 0L;
+ private long triggerTimestamp = 0L;
+
+ /** Savepoint trigger mechanism. */
+ private SavepointTriggerType triggerType = SavepointTriggerType.UNKNOWN;
/** List of recent savepoints. */
private List<Savepoint> savepointHistory = new ArrayList<>();
- public void setTrigger(String triggerId) {
+ /** Trigger timestamp of last periodic savepoint operation. */
+ private long lastPeriodicSavepointTimestamp = 0L;
+
+ public void setTrigger(String triggerId, SavepointTriggerType triggerType)
{
this.triggerId = triggerId;
this.triggerTimestamp = System.currentTimeMillis();
+ this.triggerType = triggerType;
}
- public boolean resetTrigger() {
- boolean reseted = StringUtils.isNotEmpty(this.triggerId);
+ public void resetTrigger() {
this.triggerId = "";
this.triggerTimestamp = 0L;
- return reseted;
- }
-
- public void updateLastSavepoint(Savepoint savepoint) {
- lastSavepoint = savepoint;
- resetTrigger();
}
/**
- * Add the savepoint to the history if it isn't already the most recent
savepoint.
+ * Update last savepoint info and add the savepoint to the history if it
isn't already the most
+ * recent savepoint.
*
- * @param newSavepoint
+ * @param savepoint
*/
- public void addSavepointToHistory(Savepoint newSavepoint) {
- if (!savepointHistory.isEmpty()) {
- Savepoint recentSp = savepointHistory.get(savepointHistory.size()
- 1);
- if (recentSp.getLocation().equals(newSavepoint.getLocation())) {
- return;
+ public void updateLastSavepoint(Savepoint savepoint) {
+ if (lastSavepoint == null ||
!lastSavepoint.getLocation().equals(savepoint.getLocation())) {
+ lastSavepoint = savepoint;
+ savepointHistory.add(savepoint);
+ if (savepoint.getTriggerType() == SavepointTriggerType.PERIODIC) {
+ lastPeriodicSavepointTimestamp = savepoint.getTimeStamp();
}
}
- savepointHistory.add(newSavepoint);
+ resetTrigger();
}
}
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/SavepointTriggerType.java
similarity index 58%
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/SavepointTriggerType.java
index 73da6da..d921aaa 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/SavepointTriggerType.java
@@ -17,25 +17,14 @@
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. */
- private long timeStamp;
-
- /** External pointer of the savepoint can be used to recover jobs. */
- private String location;
-
- public static Savepoint of(String location) {
- return new Savepoint(System.currentTimeMillis(), location);
- }
+/** Savepoint trigger mechanism. */
+public enum SavepointTriggerType {
+ /** Savepoint manually triggered by changing the savepointTriggerNonce. */
+ MANUAL,
+ /** Savepoint periodically triggered by the operator. */
+ PERIODIC,
+ /** Savepoint triggered during stateful upgrade. */
+ UPGRADE,
+ /** Savepoint trigger mechanism unknown, such as savepoint retrieved
directly from Flink job. */
+ UNKNOWN
}
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserver.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserver.java
index 7eecb24..30df728 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserver.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserver.java
@@ -55,10 +55,10 @@ public abstract class JobStatusObserver<CTX> {
clusterJobStatuses = new
ArrayList<>(flinkService.listJobs(deployedConfig));
} catch (Exception e) {
LOG.error("Exception while listing jobs", e);
+ ifRunningMoveToReconciling(jobStatus, previousJobStatus);
if (e instanceof TimeoutException) {
onTimeout(ctx);
}
- ifRunningMoveToReconciling(jobStatus, previousJobStatus);
return false;
}
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/SavepointFetchResult.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/SavepointFetchResult.java
index 9f111fa..813c3b7 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/SavepointFetchResult.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/SavepointFetchResult.java
@@ -17,14 +17,12 @@
package org.apache.flink.kubernetes.operator.observer;
-import org.apache.flink.kubernetes.operator.crd.status.Savepoint;
-
import lombok.Value;
/** Result of a fetch savepoint operation. */
@Value
public class SavepointFetchResult {
- Savepoint savepoint;
+ String location;
boolean pending;
String error;
@@ -36,7 +34,7 @@ public class SavepointFetchResult {
return new SavepointFetchResult(null, true, null);
}
- public static SavepointFetchResult completed(Savepoint savepoint) {
- return new SavepointFetchResult(savepoint, false, null);
+ public static SavepointFetchResult completed(String location) {
+ return new SavepointFetchResult(location, false, null);
}
}
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 91726d7..a0b5788 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
@@ -77,10 +77,11 @@ public class SavepointObserver<STATUS extends
CommonStatus<?>> {
resource,
EventUtils.Type.Warning,
"SavepointError",
- "Savepoint failed for
savepointTriggerNonce: "
- + resource.getSpec()
+ SavepointUtils.createSavepointError(
+ savepointInfo,
+ resource.getSpec()
.getJob()
-
.getSavepointTriggerNonce(),
+
.getSavepointTriggerNonce()),
EventUtils.Component.Operator));
// We only need to observe latest checkpoint/savepoint for terminal
jobs
@@ -145,10 +146,14 @@ public class SavepointObserver<STATUS extends
CommonStatus<?>> {
}
LOG.info("Savepoint status updated with latest completed savepoint
info");
-
currentSavepointInfo.updateLastSavepoint(savepointFetchResult.getSavepoint());
-
- updateSavepointHistory(
- currentSavepointInfo, savepointFetchResult.getSavepoint(),
deployedConfig);
+ var savepoint =
+ new Savepoint(
+ currentSavepointInfo.getTriggerTimestamp(),
+ savepointFetchResult.getLocation(),
+ currentSavepointInfo.getTriggerType());
+ currentSavepointInfo.updateLastSavepoint(savepoint);
+
+ updateSavepointHistory(currentSavepointInfo, savepoint,
deployedConfig);
return Optional.empty();
}
@@ -158,8 +163,6 @@ public class SavepointObserver<STATUS extends
CommonStatus<?>> {
Savepoint newSavepoint,
Configuration deployedConfig) {
- currentSavepointInfo.addSavepointToHistory(newSavepoint);
-
// maintain history
List<Savepoint> savepointHistory =
currentSavepointInfo.getSavepointHistory();
int maxCount =
configManager.getOperatorConfiguration().getSavepointHistoryMaxCount();
@@ -195,11 +198,7 @@ public class SavepointObserver<STATUS extends
CommonStatus<?>> {
try {
flinkService
.getLastCheckpoint(JobID.fromHexString(jobID),
deployedConfig)
- .ifPresent(
- sp -> {
- savepointInfo.updateLastSavepoint(sp);
- savepointInfo.addSavepointToHistory(sp);
- });
+ .ifPresent(savepointInfo::updateLastSavepoint);
} catch (Exception e) {
LOG.error("Could not observe latest savepoint information.", e);
throw new ReconciliationException(e);
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/AbstractDeploymentObserver.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/AbstractDeploymentObserver.java
index 8590f16..c17647f 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/AbstractDeploymentObserver.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/AbstractDeploymentObserver.java
@@ -33,6 +33,7 @@ import org.apache.flink.kubernetes.operator.observer.Observer;
import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
import org.apache.flink.kubernetes.operator.service.FlinkService;
import org.apache.flink.kubernetes.operator.utils.EventUtils;
+import org.apache.flink.kubernetes.operator.utils.SavepointUtils;
import io.fabric8.kubernetes.api.model.ContainerStateWaiting;
import io.fabric8.kubernetes.api.model.ContainerStatus;
@@ -72,10 +73,9 @@ public abstract class AbstractDeploymentObserver implements
Observer<FlinkDeploy
@Override
public void observe(FlinkDeployment flinkApp, Context context) {
- ReconciliationStatus<FlinkDeploymentSpec> reconciliationStatus =
- flinkApp.getStatus().getReconciliationStatus();
- FlinkDeploymentSpec lastReconciledSpec =
- reconciliationStatus.deserializeLastReconciledSpec();
+ var status = flinkApp.getStatus();
+ var reconciliationStatus = status.getReconciliationStatus();
+ var lastReconciledSpec =
reconciliationStatus.deserializeLastReconciledSpec();
// Nothing has been launched so skip observing
if (lastReconciledSpec == null
@@ -101,20 +101,7 @@ public abstract class AbstractDeploymentObserver
implements Observer<FlinkDeploy
observeClusterInfo(flinkApp, observeConfig);
}
- if (!ReconciliationUtils.isJobRunning(flinkApp.getStatus())) {
- if
(flinkApp.getStatus().getJobStatus().getSavepointInfo().resetTrigger()) {
- logger.error("Job is not running, cancelling savepoint
operation");
- EventUtils.createOrUpdateEvent(
- flinkService.getKubernetesClient(),
- flinkApp,
- EventUtils.Type.Warning,
- "SavepointError",
- "Savepoint failed for savepointTriggerNonce: "
- +
flinkApp.getSpec().getJob().getSavepointTriggerNonce(),
- EventUtils.Component.Operator);
- }
- }
-
+
SavepointUtils.resetTriggerIfJobNotRunning(flinkService.getKubernetesClient(),
flinkApp);
clearErrorsIfDeploymentIsHealthy(flinkApp);
}
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/sessionjob/SessionJobObserver.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/sessionjob/SessionJobObserver.java
index d1146e9..a0f6c73 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/sessionjob/SessionJobObserver.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/sessionjob/SessionJobObserver.java
@@ -28,6 +28,7 @@ import
org.apache.flink.kubernetes.operator.observer.SavepointObserver;
import
org.apache.flink.kubernetes.operator.observer.context.VoidObserverContext;
import
org.apache.flink.kubernetes.operator.reconciler.sessionjob.SessionJobHelper;
import org.apache.flink.kubernetes.operator.service.FlinkService;
+import org.apache.flink.kubernetes.operator.utils.SavepointUtils;
import org.apache.flink.kubernetes.operator.utils.StatusHelper;
import org.apache.flink.runtime.client.JobStatusMessage;
import org.apache.flink.util.Preconditions;
@@ -44,6 +45,7 @@ import java.util.stream.Collectors;
public class SessionJobObserver implements Observer<FlinkSessionJob> {
private static final Logger LOG =
LoggerFactory.getLogger(SessionJobObserver.class);
+ private final FlinkService flinkService;
private final FlinkConfigManager configManager;
private final SavepointObserver savepointObserver;
private final JobStatusObserver<VoidObserverContext> jobStatusObserver;
@@ -52,6 +54,7 @@ public class SessionJobObserver implements
Observer<FlinkSessionJob> {
FlinkService flinkService,
FlinkConfigManager configManager,
StatusHelper<FlinkSessionJobStatus> statusHelper) {
+ this.flinkService = flinkService;
this.configManager = configManager;
this.savepointObserver = new SavepointObserver(flinkService,
configManager, statusHelper);
this.jobStatusObserver =
@@ -118,5 +121,7 @@ public class SessionJobObserver implements
Observer<FlinkSessionJob> {
if (jobFound) {
savepointObserver.observeSavepointStatus(flinkSessionJob,
deployedConfig);
}
+ SavepointUtils.resetTriggerIfJobNotRunning(
+ flinkService.getKubernetesClient(), flinkSessionJob);
}
}
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 8adfafa..4a21170 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
@@ -286,17 +286,15 @@ public class ReconciliationUtils {
== JobManagerDeploymentStatus.MISSING);
}
- public static boolean isJobInTerminalState(FlinkDeploymentStatus status) {
+ public static boolean isJobInTerminalState(CommonStatus<?> status) {
var jobState = status.getJobStatus().getState();
return
org.apache.flink.api.common.JobStatus.valueOf(jobState).isGloballyTerminalState();
}
- public static boolean isJobRunning(FlinkDeploymentStatus status) {
- JobManagerDeploymentStatus deploymentStatus =
status.getJobManagerDeploymentStatus();
- return deploymentStatus == JobManagerDeploymentStatus.READY
- && org.apache.flink.api.common.JobStatus.RUNNING
- .name()
- .equals(status.getJobStatus().getState());
+ public static boolean isJobRunning(CommonStatus<?> status) {
+ return org.apache.flink.api.common.JobStatus.RUNNING
+ .name()
+ .equals(status.getJobStatus().getState());
}
/**
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 442206f..18551fa 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
@@ -142,12 +142,10 @@ public class ApplicationReconciler extends
AbstractDeploymentReconciler {
rollbackApplication(flinkApp);
} else if (ReconciliationUtils.shouldRecoverDeployment(observeConfig,
flinkApp)) {
recoverJmDeployment(flinkApp, observeConfig);
- } else if (SavepointUtils.shouldTriggerSavepoint(desiredJobSpec,
status)
- && ReconciliationUtils.isJobRunning(status)) {
- triggerSavepoint(flinkApp);
- ReconciliationUtils.updateSavepointReconciliationSuccess(flinkApp);
} else {
- LOG.info("Deployment is fully reconciled, nothing to do.");
+ if (!SavepointUtils.triggerSavepointIfNeeded(flinkService,
flinkApp, observeConfig)) {
+ LOG.info("Deployment is fully reconciled, nothing to do.");
+ }
}
}
@@ -326,11 +324,4 @@ public class ApplicationReconciler extends
AbstractDeploymentReconciler {
protected void shutdown(FlinkDeployment flinkApp) {
flinkService.cancelJob(flinkApp, UpgradeMode.STATELESS);
}
-
- private void triggerSavepoint(FlinkDeployment deployment) throws Exception
{
- flinkService.triggerSavepoint(
- deployment.getStatus().getJobStatus().getJobId(),
- deployment.getStatus().getJobStatus().getSavepointInfo(),
- configManager.getObserveConfig(deployment));
- }
}
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 1b1eeb4..c46f3e3 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
@@ -28,9 +28,11 @@ 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.crd.status.Savepoint;
+import org.apache.flink.kubernetes.operator.crd.status.SavepointTriggerType;
import org.apache.flink.kubernetes.operator.reconciler.Reconciler;
import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
import org.apache.flink.kubernetes.operator.service.FlinkService;
+import org.apache.flink.kubernetes.operator.utils.SavepointUtils;
import org.apache.flink.util.Preconditions;
import io.fabric8.kubernetes.client.KubernetesClient;
@@ -79,13 +81,13 @@ public class FlinkSessionJobReconciler implements
Reconciler<FlinkSessionJob> {
return;
}
- Configuration deployedConfig =
+ Configuration sessionJobConfig =
configManager.getSessionJobConfig(flinkDepOptional.get(),
flinkSessionJob);
if (lastReconciledSpec == null) {
submitAndInitStatus(
flinkSessionJob,
- deployedConfig,
+ sessionJobConfig,
Optional.ofNullable(
flinkSessionJob.getSpec().getJob().getInitialSavepointPath())
.orElse(null));
@@ -94,17 +96,17 @@ public class FlinkSessionJobReconciler implements
Reconciler<FlinkSessionJob> {
return;
}
- if (!deployedConfig.getBoolean(
+ if (!sessionJobConfig.getBoolean(
KubernetesOperatorConfigOptions.JOB_UPGRADE_IGNORE_PENDING_SAVEPOINT)
- && helper.savepointInProgress()) {
+ &&
SavepointUtils.savepointInProgress(flinkSessionJob.getStatus().getJobStatus()))
{
LOG.info("Delaying job reconciliation until pending savepoint is
completed");
return;
}
boolean specChanged = helper.specChanged(lastReconciledSpec);
+ var jobSpec = flinkSessionJob.getSpec().getJob();
if (specChanged) {
- var jobSpec = flinkSessionJob.getSpec().getJob();
JobState currentJobState = lastReconciledSpec.getJob().getState();
JobState desiredJobState = jobSpec.getState();
UpgradeMode upgradeMode = jobSpec.getUpgradeMode();
@@ -113,22 +115,24 @@ public class FlinkSessionJobReconciler implements
Reconciler<FlinkSessionJob> {
if (desiredJobState == JobState.RUNNING) {
LOG.info("Upgrading/Restarting running job, suspending
first...");
}
- stateAfterReconcile = suspendJob(flinkSessionJob, upgradeMode,
deployedConfig);
+ stateAfterReconcile = suspendJob(flinkSessionJob, upgradeMode,
sessionJobConfig);
}
if (currentJobState == JobState.SUSPENDED && desiredJobState ==
JobState.RUNNING) {
if (upgradeMode == UpgradeMode.STATELESS) {
- submitAndInitStatus(flinkSessionJob, deployedConfig, null);
+ submitAndInitStatus(flinkSessionJob, sessionJobConfig,
null);
} else if (upgradeMode == UpgradeMode.LAST_STATE
|| upgradeMode == UpgradeMode.SAVEPOINT) {
- restoreFromLastSavepoint(flinkSessionJob, deployedConfig);
+ restoreFromLastSavepoint(flinkSessionJob,
sessionJobConfig);
}
stateAfterReconcile = JobState.RUNNING;
}
ReconciliationUtils.updateForSpecReconciliationSuccess(
flinkSessionJob, stateAfterReconcile);
- } else if (helper.shouldTriggerSavepoint() &&
helper.isJobRunning(flinkDepOptional.get())) {
- triggerSavepoint(flinkSessionJob, deployedConfig);
-
ReconciliationUtils.updateSavepointReconciliationSuccess(flinkSessionJob);
+ } else {
+ if (!SavepointUtils.triggerSavepointIfNeeded(
+ flinkService, flinkSessionJob, sessionJobConfig)) {
+ LOG.info("Session Job is fully reconciled, nothing to do.");
+ }
}
}
@@ -199,18 +203,9 @@ public class FlinkSessionJobReconciler implements
Reconciler<FlinkSessionJob> {
jobStatus.setState(stateAfterReconcile.name());
savepointOpt.ifPresent(
location -> {
- Savepoint sp = Savepoint.of(location);
- jobStatus.getSavepointInfo().setLastSavepoint(sp);
- jobStatus.getSavepointInfo().addSavepointToHistory(sp);
+ Savepoint sp = Savepoint.of(location,
SavepointTriggerType.UPGRADE);
+ jobStatus.getSavepointInfo().updateLastSavepoint(sp);
});
return stateAfterReconcile;
}
-
- private void triggerSavepoint(FlinkSessionJob sessionJob, Configuration
effectiveConfig)
- throws Exception {
- flinkService.triggerSavepoint(
- sessionJob.getStatus().getJobStatus().getJobId(),
- sessionJob.getStatus().getJobStatus().getSavepointInfo(),
- effectiveConfig);
- }
}
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobHelper.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobHelper.java
index bca3789..4af66ac 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobHelper.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/sessionjob/SessionJobHelper.java
@@ -20,9 +20,7 @@ package
org.apache.flink.kubernetes.operator.reconciler.sessionjob;
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;
-import org.apache.flink.kubernetes.operator.crd.status.FlinkDeploymentStatus;
import
org.apache.flink.kubernetes.operator.crd.status.JobManagerDeploymentStatus;
-import org.apache.flink.kubernetes.operator.utils.SavepointUtils;
import org.slf4j.Logger;
@@ -33,48 +31,16 @@ public class SessionJobHelper {
private final Logger logger;
private final FlinkSessionJob sessionJob;
- private final FlinkSessionJobSpec lastReconciledSpec;
public SessionJobHelper(FlinkSessionJob sessionJob, Logger logger) {
this.sessionJob = sessionJob;
- this.lastReconciledSpec =
-
sessionJob.getStatus().getReconciliationStatus().deserializeLastReconciledSpec();
this.logger = logger;
}
- public boolean savepointInProgress() {
- return
SavepointUtils.savepointInProgress(sessionJob.getStatus().getJobStatus());
- }
-
- public boolean shouldTriggerSavepoint() {
- if (lastReconciledSpec == null) {
- return false;
- }
- if (savepointInProgress()) {
- return false;
- }
- return savepointTriggerNonce() != null
- && !savepointTriggerNonce()
-
.equals(lastReconciledSpec.getJob().getSavepointTriggerNonce());
- }
-
- private Long savepointTriggerNonce() {
- return sessionJob.getSpec().getJob().getSavepointTriggerNonce();
- }
-
public boolean specChanged(FlinkSessionJobSpec lastReconciledSpec) {
return !sessionJob.getSpec().equals(lastReconciledSpec);
}
- public boolean isJobRunning(FlinkDeployment deployment) {
- FlinkDeploymentStatus status = deployment.getStatus();
- JobManagerDeploymentStatus deploymentStatus =
status.getJobManagerDeploymentStatus();
- return deploymentStatus == JobManagerDeploymentStatus.READY
- && org.apache.flink.api.common.JobStatus.RUNNING
- .name()
-
.equals(sessionJob.getStatus().getJobStatus().getState());
- }
-
public boolean sessionClusterReady(Optional<FlinkDeployment>
flinkDeploymentOpt) {
if (flinkDeploymentOpt.isPresent()) {
var flinkdep = flinkDeploymentOpt.get();
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 304748e..c6731c0 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
@@ -45,6 +45,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.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;
@@ -60,6 +61,8 @@ import
org.apache.flink.runtime.rest.messages.DashboardConfiguration;
import org.apache.flink.runtime.rest.messages.EmptyMessageParameters;
import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
import org.apache.flink.runtime.rest.messages.TriggerId;
+import
org.apache.flink.runtime.rest.messages.job.savepoints.SavepointDisposalRequest;
+import
org.apache.flink.runtime.rest.messages.job.savepoints.SavepointDisposalTriggerHeaders;
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;
@@ -434,10 +437,8 @@ public class FlinkService {
deploymentStatus.getJobStatus().setState(JobStatus.FINISHED.name());
savepointOpt.ifPresent(
location -> {
- Savepoint sp = Savepoint.of(location);
-
deploymentStatus.getJobStatus().getSavepointInfo().setLastSavepoint(sp);
- // this is required for Flink < 1.15
-
deploymentStatus.getJobStatus().getSavepointInfo().addSavepointToHistory(sp);
+ Savepoint sp = Savepoint.of(location,
SavepointTriggerType.UPGRADE);
+
deploymentStatus.getJobStatus().getSavepointInfo().updateLastSavepoint(sp);
});
var shutdownDisabled =
@@ -536,6 +537,7 @@ public class FlinkService {
public void triggerSavepoint(
String jobId,
+ SavepointTriggerType triggerType,
org.apache.flink.kubernetes.operator.crd.status.SavepointInfo
savepointInfo,
Configuration conf)
throws Exception {
@@ -567,8 +569,7 @@ public class FlinkService {
.get(timeout, TimeUnit.SECONDS);
LOG.info("Savepoint successfully triggered: " +
response.getTriggerId().toHexString());
- savepointInfo.setTrigger(response.getTriggerId().toHexString());
- savepointInfo.setTriggerTimestamp(System.currentTimeMillis());
+ savepointInfo.setTrigger(response.getTriggerId().toHexString(),
triggerType);
}
}
@@ -603,7 +604,8 @@ public class FlinkService {
"Latest checkpoint not externally addressable, manual
recovery required.",
"CheckpointNotFound");
}
- return latestCheckpointOpt.map(Savepoint::of);
+ return latestCheckpointOpt.map(
+ pointer -> Savepoint.of(pointer,
SavepointTriggerType.UNKNOWN));
}
}
@@ -639,11 +641,9 @@ public class FlinkService {
return SavepointFetchResult.pending();
}
}
- Savepoint savepoint =
- new Savepoint(
- System.currentTimeMillis(),
response.get().resource().getLocation());
- LOG.info("Savepoint result: " + savepoint);
- return SavepointFetchResult.completed(savepoint);
+ String location = response.get().resource().getLocation();
+ LOG.info("Savepoint result: {}", location);
+ return SavepointFetchResult.completed(location);
} catch (Exception e) {
LOG.error("Exception while fetching the savepoint result", e);
return SavepointFetchResult.error(e.getMessage());
@@ -653,7 +653,17 @@ public class FlinkService {
public void disposeSavepoint(String savepointPath, Configuration conf)
throws Exception {
try (RestClusterClient<String> clusterClient =
(RestClusterClient<String>) getClusterClient(conf)) {
- clusterClient.disposeSavepoint(savepointPath);
+ clusterClient
+ .sendRequest(
+ SavepointDisposalTriggerHeaders.getInstance(),
+ EmptyMessageParameters.getInstance(),
+ new SavepointDisposalRequest(savepointPath))
+ .get(
+ configManager
+ .getOperatorConfiguration()
+ .getFlinkClientTimeout()
+ .getSeconds(),
+ TimeUnit.SECONDS);
}
}
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 8383f32..c4ca34e 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
@@ -17,40 +17,153 @@
package org.apache.flink.kubernetes.operator.utils;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.kubernetes.operator.config.FlinkOperatorConfiguration;
-import org.apache.flink.kubernetes.operator.crd.spec.JobSpec;
-import org.apache.flink.kubernetes.operator.crd.status.FlinkDeploymentStatus;
+import
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
+import org.apache.flink.kubernetes.operator.crd.AbstractFlinkResource;
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;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.service.FlinkService;
+import io.fabric8.kubernetes.client.KubernetesClient;
import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.time.Duration;
+import java.time.Instant;
+import java.util.Optional;
/** Savepoint utilities. */
public class SavepointUtils {
+ private static final Logger LOG =
LoggerFactory.getLogger(SavepointUtils.class);
+
public static boolean savepointInProgress(JobStatus jobStatus) {
return
StringUtils.isNotEmpty(jobStatus.getSavepointInfo().getTriggerId());
}
- public static boolean shouldTriggerSavepoint(JobSpec jobSpec,
FlinkDeploymentStatus status) {
- if (savepointInProgress(status.getJobStatus())) {
+ /**
+ * Triggers any pending manual or periodic savepoints and updates the
status accordingly.
+ *
+ * @param flinkService {@link FlinkService} used to trigger savepoints
+ * @param resource Resource that should be savepointed
+ * @param conf Observe config of the resource
+ * @return True if a savepoint was triggered
+ */
+ public static boolean triggerSavepointIfNeeded(
+ FlinkService flinkService, AbstractFlinkResource<?, ?> resource,
Configuration conf)
+ throws Exception {
+
+ Optional<SavepointTriggerType> triggerOpt =
shouldTriggerSavepoint(resource, conf);
+ if (triggerOpt.isEmpty()) {
return false;
}
- return jobSpec.getSavepointTriggerNonce() != null
- && !jobSpec.getSavepointTriggerNonce()
- .equals(
- status.getReconciliationStatus()
- .deserializeLastReconciledSpec()
- .getJob()
- .getSavepointTriggerNonce());
+
+ var triggerType = triggerOpt.get();
+ flinkService.triggerSavepoint(
+ resource.getStatus().getJobStatus().getJobId(),
+ triggerType,
+ resource.getStatus().getJobStatus().getSavepointInfo(),
+ conf);
+
+ if (triggerType == SavepointTriggerType.MANUAL) {
+ ReconciliationUtils.updateSavepointReconciliationSuccess(resource);
+ }
+ return true;
+ }
+
+ /**
+ * Checks whether savepoint should be triggered based on the current
status and spec and if yes,
+ * returns the correct {@link SavepointTriggerType}.
+ *
+ * <p>This logic is responsible for both manual and periodic savepoint
triggering.
+ *
+ * @param resource Resource to be savepointed
+ * @param conf Observe configuration of the resource
+ * @return Optional @{@link SavepointTriggerType}
+ */
+ @VisibleForTesting
+ protected static Optional<SavepointTriggerType> shouldTriggerSavepoint(
+ AbstractFlinkResource<?, ?> resource, Configuration conf) {
+
+ var status = resource.getStatus();
+ var jobSpec = resource.getSpec().getJob();
+ var jobStatus = status.getJobStatus();
+
+ if (!ReconciliationUtils.isJobRunning(status) ||
savepointInProgress(jobStatus)) {
+ return Optional.empty();
+ }
+
+ var triggerNonceChanged =
+ jobSpec.getSavepointTriggerNonce() != null
+ && !jobSpec.getSavepointTriggerNonce()
+ .equals(
+ status.getReconciliationStatus()
+
.deserializeLastReconciledSpec()
+ .getJob()
+ .getSavepointTriggerNonce());
+ if (triggerNonceChanged) {
+ return Optional.of(SavepointTriggerType.MANUAL);
+ }
+
+ var savepointInterval =
+
conf.get(KubernetesOperatorConfigOptions.PERIODIC_SAVEPOINT_INTERVAL);
+
+ if (savepointInterval.isZero()) {
+ return Optional.empty();
+ }
+
+ var lastTriggerTs =
jobStatus.getSavepointInfo().getLastPeriodicSavepointTimestamp();
+
+ // When the resource is first created/periodic savepointing enabled we
have to compare
+ // against the creation timestamp for triggering the first periodic
savepoint
+ var lastTrigger =
+ lastTriggerTs == 0
+ ?
Instant.parse(resource.getMetadata().getCreationTimestamp())
+ : Instant.ofEpochMilli(lastTriggerTs);
+ var now = Instant.now();
+ if (lastTrigger.plus(savepointInterval).isBefore(Instant.now())) {
+ LOG.info(
+ "Triggering new periodic savepoint after {}",
+ Duration.between(lastTrigger, now));
+ return Optional.of(SavepointTriggerType.PERIODIC);
+ }
+ return Optional.empty();
}
public static boolean gracePeriodEnded(
FlinkOperatorConfiguration configuration, SavepointInfo
savepointInfo) {
- Duration gracePeriod = configuration.getSavepointTriggerGracePeriod();
- long triggerTimestamp = savepointInfo.getTriggerTimestamp();
- return (System.currentTimeMillis() - triggerTimestamp) >
gracePeriod.toMillis();
+ var elapsed = System.currentTimeMillis() -
savepointInfo.getTriggerTimestamp();
+ return elapsed >
configuration.getSavepointTriggerGracePeriod().toMillis();
+ }
+
+ public static void resetTriggerIfJobNotRunning(
+ KubernetesClient client, AbstractFlinkResource<?, ?> resource) {
+ var status = resource.getStatus();
+ var jobStatus = status.getJobStatus();
+ if (!ReconciliationUtils.isJobRunning(status)
+ && SavepointUtils.savepointInProgress(jobStatus)) {
+ var savepointInfo = jobStatus.getSavepointInfo();
+ savepointInfo.resetTrigger();
+ LOG.error("Job is not running, cancelling savepoint operation");
+ EventUtils.createOrUpdateEvent(
+ client,
+ resource,
+ EventUtils.Type.Warning,
+ "SavepointError",
+ createSavepointError(
+ savepointInfo,
resource.getSpec().getJob().getSavepointTriggerNonce()),
+ EventUtils.Component.Operator);
+ }
+ }
+
+ public static String createSavepointError(SavepointInfo savepointInfo,
Long triggerNonce) {
+ return SavepointTriggerType.PERIODIC == savepointInfo.getTriggerType()
+ ? "Periodic savepoint failed"
+ : "Savepoint failed for savepointTriggerNonce: " +
triggerNonce;
}
}
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java
index f04d0f8..d4afbe4 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java
@@ -197,6 +197,12 @@ public class DefaultValidator implements
FlinkResourceValidator {
String.format(
"Savepoint could not be manually triggered for
the running job while config key[%s] is not set",
CheckpointingOptions.SAVEPOINT_DIRECTORY.key()));
+ } else if (configuration.contains(
+
KubernetesOperatorConfigOptions.PERIODIC_SAVEPOINT_INTERVAL)) {
+ return Optional.of(
+ String.format(
+ "Periodic savepoints cannot be enabled when
config key[%s] is not set",
+
CheckpointingOptions.SAVEPOINT_DIRECTORY.key()));
}
}
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestUtils.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestUtils.java
index 846d641..306eb3a 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestUtils.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestUtils.java
@@ -70,6 +70,7 @@ import org.junit.jupiter.api.Assertions;
import java.lang.reflect.Field;
import java.net.HttpURLConnection;
+import java.time.Instant;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@@ -102,6 +103,7 @@ public class TestUtils {
new ObjectMetaBuilder()
.withName(TEST_DEPLOYMENT_NAME)
.withNamespace(TEST_NAMESPACE)
+ .withCreationTimestamp(Instant.now().toString())
.build());
deployment.setSpec(getTestFlinkDeploymentSpec(version));
return deployment;
@@ -132,6 +134,7 @@ public class TestUtils {
new ObjectMetaBuilder()
.withName(TEST_SESSION_JOB_NAME)
.withNamespace(TEST_NAMESPACE)
+ .withCreationTimestamp(Instant.now().toString())
.build());
sessionJob.setSpec(
FlinkSessionJobSpec.builder()
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 27904f4..ac21037 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
@@ -35,6 +35,8 @@ 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.SavepointInfo;
+import org.apache.flink.kubernetes.operator.crd.status.SavepointTriggerType;
import org.apache.flink.kubernetes.operator.observer.SavepointFetchResult;
import org.apache.flink.kubernetes.operator.service.FlinkService;
import org.apache.flink.kubernetes.operator.utils.FlinkUtils;
@@ -88,7 +90,7 @@ public class TestingFlinkService extends FlinkService {
private PodList podList = new PodList();
private Consumer<Configuration> listJobConsumer = conf -> {};
private List<String> disposedSavepoints = new ArrayList<>();
- private SavepointFetchResult savepointFetchResult;
+ private Map<String, Boolean> savepointTriggers = new HashMap<>();
public TestingFlinkService() {
super(null, new FlinkConfigManager(new Configuration()));
@@ -234,10 +236,27 @@ public class TestingFlinkService extends FlinkService {
@Override
public void triggerSavepoint(
String jobId,
- org.apache.flink.kubernetes.operator.crd.status.SavepointInfo
savepointInfo,
- Configuration conf)
- throws Exception {
- savepointInfo.setTrigger("trigger_" + triggerCounter++);
+ SavepointTriggerType triggerType,
+ SavepointInfo savepointInfo,
+ Configuration conf) {
+ var triggerId = "trigger_" + triggerCounter++;
+ savepointInfo.setTrigger(triggerId, triggerType);
+ savepointTriggers.put(triggerId, false);
+ }
+
+ @Override
+ public SavepointFetchResult fetchSavepointInfo(
+ String triggerId, String jobId, Configuration conf) {
+
+ if (savepointTriggers.containsKey(triggerId)) {
+ if (savepointTriggers.get(triggerId)) {
+ return SavepointFetchResult.completed("savepoint_" +
savepointCounter++);
+ }
+ savepointTriggers.put(triggerId, true);
+ return SavepointFetchResult.pending();
+ }
+
+ return SavepointFetchResult.error("Failed");
}
@Override
@@ -322,23 +341,6 @@ public class TestingFlinkService extends FlinkService {
@Override
protected void waitForClusterShutdown(Configuration conf) {}
- @Override
- public SavepointFetchResult fetchSavepointInfo(
- String triggerId, String jobId, Configuration conf) {
-
- if (savepointFetchResult == null) {
- savepointFetchResult = SavepointFetchResult.pending();
- return savepointFetchResult;
- }
-
- if (savepointFetchResult.isPending()) {
- savepointFetchResult = SavepointFetchResult.error("Failed");
- return savepointFetchResult;
- }
-
- return SavepointFetchResult.completed(Savepoint.of("savepoint_" +
savepointCounter++));
- }
-
@Override
public void disposeSavepoint(String savepointPath, Configuration conf) {
disposedSavepoints.add(savepointPath);
@@ -363,7 +365,7 @@ public class TestingFlinkService extends FlinkService {
}
if (t.f0 != null) {
- return Optional.of(Savepoint.of(t.f0));
+ return Optional.of(Savepoint.of(t.f0,
SavepointTriggerType.UNKNOWN));
} else {
return Optional.empty();
}
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/RollbackTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/RollbackTest.java
index 7ea199f..4774e3b 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/RollbackTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/RollbackTest.java
@@ -30,6 +30,7 @@ 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.ReconciliationState;
import org.apache.flink.kubernetes.operator.crd.status.Savepoint;
+import org.apache.flink.kubernetes.operator.crd.status.SavepointTriggerType;
import org.apache.flink.util.function.ThrowingRunnable;
import io.fabric8.kubernetes.client.KubernetesClient;
@@ -214,7 +215,8 @@ public class RollbackTest {
dep.getStatus()
.getJobStatus()
.getSavepointInfo()
- .updateLastSavepoint(Savepoint.of("test"));
+ .updateLastSavepoint(
+ Savepoint.of("test",
SavepointTriggerType.UPGRADE));
testController.reconcile(dep, context);
},
() -> {
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 c5d18c4..e3ff8d0 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
@@ -18,12 +18,15 @@
package org.apache.flink.kubernetes.operator.observer;
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.TestingStatusHelper;
import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
import
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
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.utils.SavepointUtils;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -31,6 +34,10 @@ import org.junit.jupiter.api.Test;
import java.time.Duration;
import java.util.Collections;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
/** Tests for {@link SavepointObserver}. */
public class SavepointObserverTest {
@@ -44,7 +51,8 @@ public class SavepointObserverTest {
SavepointInfo spInfo = new SavepointInfo();
Assertions.assertTrue(spInfo.getSavepointHistory().isEmpty());
- Savepoint sp = new Savepoint(1, "sp1");
+ Savepoint sp = new Savepoint(1, "sp1", SavepointTriggerType.MANUAL);
+ spInfo.updateLastSavepoint(sp);
observer.updateSavepointHistory(spInfo, sp,
configManager.getDefaultConfig());
Assertions.assertNotNull(spInfo.getSavepointHistory());
@@ -63,18 +71,54 @@ public class SavepointObserverTest {
new SavepointObserver(flinkService, configManager, new
TestingStatusHelper<>());
SavepointInfo spInfo = new SavepointInfo();
- Savepoint sp1 = new Savepoint(1, "sp1");
+ Savepoint sp1 = new Savepoint(1, "sp1", SavepointTriggerType.MANUAL);
+ spInfo.updateLastSavepoint(sp1);
observer.updateSavepointHistory(spInfo, sp1, conf);
Assertions.assertIterableEquals(
Collections.singletonList(sp1), spInfo.getSavepointHistory());
Assertions.assertIterableEquals(
Collections.emptyList(), flinkService.getDisposedSavepoints());
- Savepoint sp2 = new Savepoint(2, "sp2");
+ Savepoint sp2 = new Savepoint(2, "sp2", SavepointTriggerType.MANUAL);
+ spInfo.updateLastSavepoint(sp2);
observer.updateSavepointHistory(spInfo, sp2, conf);
Assertions.assertIterableEquals(
Collections.singletonList(sp2), spInfo.getSavepointHistory());
Assertions.assertIterableEquals(
Collections.singletonList(sp1.getLocation()),
flinkService.getDisposedSavepoints());
}
+
+ @Test
+ public void testPeriodicSavepoint() throws Exception {
+ var conf = new Configuration();
+ var deployment = TestUtils.buildApplicationCluster();
+ deployment
+ .getStatus()
+ .getReconciliationStatus()
+ .serializeAndSetLastReconciledSpec(deployment.getSpec());
+
+ var jobStatus = deployment.getStatus().getJobStatus();
+ jobStatus.setState("RUNNING");
+
+ var savepointInfo = jobStatus.getSavepointInfo();
+ var observer =
+ new SavepointObserver(flinkService, configManager, new
TestingStatusHelper<>());
+ flinkService.triggerSavepoint(null, SavepointTriggerType.PERIODIC,
savepointInfo, conf);
+
+ var triggerTs = savepointInfo.getTriggerTimestamp();
+ assertEquals(0L, savepointInfo.getLastPeriodicSavepointTimestamp());
+ assertEquals(SavepointTriggerType.PERIODIC,
savepointInfo.getTriggerType());
+ assertTrue(SavepointUtils.savepointInProgress(jobStatus));
+ assertTrue(triggerTs > 0);
+
+ // Pending
+ observer.observeSavepointStatus(deployment, conf);
+ // Completed
+ observer.observeSavepointStatus(deployment, conf);
+ assertEquals(triggerTs,
savepointInfo.getLastPeriodicSavepointTimestamp());
+ assertFalse(SavepointUtils.savepointInProgress(jobStatus));
+ assertEquals(savepointInfo.getLastSavepoint(),
savepointInfo.getSavepointHistory().get(0));
+ assertEquals(
+ SavepointTriggerType.PERIODIC,
savepointInfo.getLastSavepoint().getTriggerType());
+ }
}
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 2211216..3963b97 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
@@ -28,6 +28,7 @@ import org.apache.flink.kubernetes.operator.crd.spec.JobState;
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.JobStatus;
+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;
import org.apache.flink.kubernetes.operator.utils.SavepointUtils;
@@ -173,6 +174,7 @@ public class ApplicationObserverTest {
flinkService.triggerSavepoint(
deployment.getStatus().getJobStatus().getJobId(),
+ SavepointTriggerType.MANUAL,
deployment.getStatus().getJobStatus().getSavepointInfo(),
conf);
// pending savepoint
@@ -182,6 +184,7 @@ public class ApplicationObserverTest {
observer.observe(deployment, readyContext);
assertTrue(SavepointUtils.savepointInProgress(deployment.getStatus().getJobStatus()));
+
deployment.getStatus().getJobStatus().getSavepointInfo().setTriggerId("unknown");
// savepoint error
assertEquals(
0,
@@ -217,12 +220,15 @@ public class ApplicationObserverTest {
// savepoint success
flinkService.triggerSavepoint(
deployment.getStatus().getJobStatus().getJobId(),
+ SavepointTriggerType.MANUAL,
deployment.getStatus().getJobStatus().getSavepointInfo(),
conf);
assertEquals(
"trigger_1",
deployment.getStatus().getJobStatus().getSavepointInfo().getTriggerId());
observer.observe(deployment, readyContext);
+ observer.observe(deployment, readyContext);
+
assertEquals(
"savepoint_0",
deployment
@@ -236,6 +242,7 @@ public class ApplicationObserverTest {
// second attempt success
flinkService.triggerSavepoint(
deployment.getStatus().getJobStatus().getJobId(),
+ SavepointTriggerType.MANUAL,
deployment.getStatus().getJobStatus().getSavepointInfo(),
conf);
assertEquals(
@@ -244,6 +251,8 @@ public class ApplicationObserverTest {
assertTrue(SavepointUtils.savepointInProgress(deployment.getStatus().getJobStatus()));
observer.observe(deployment, readyContext);
+ observer.observe(deployment, readyContext);
+
assertEquals(
"savepoint_1",
deployment
@@ -257,6 +266,7 @@ public class ApplicationObserverTest {
// application failure after checkpoint trigger
flinkService.triggerSavepoint(
deployment.getStatus().getJobStatus().getJobId(),
+ SavepointTriggerType.MANUAL,
deployment.getStatus().getJobStatus().getSavepointInfo(),
conf);
assertEquals(
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 667067f..68a0c36 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
@@ -25,6 +25,7 @@ import org.apache.flink.kubernetes.operator.TestUtils;
import org.apache.flink.kubernetes.operator.TestingFlinkService;
import org.apache.flink.kubernetes.operator.TestingStatusHelper;
import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
+import org.apache.flink.kubernetes.operator.crd.status.SavepointTriggerType;
import
org.apache.flink.kubernetes.operator.reconciler.sessionjob.FlinkSessionJobReconciler;
import org.apache.flink.kubernetes.operator.utils.SavepointUtils;
@@ -152,20 +153,22 @@ public class SessionJobObserverTest {
Assertions.assertFalse(
SavepointUtils.savepointInProgress(sessionJob.getStatus().getJobStatus()));
- flinkService.triggerSavepoint(jobID, savepointInfo, new
Configuration());
+ flinkService.triggerSavepoint(
+ jobID, SavepointTriggerType.MANUAL, savepointInfo, new
Configuration());
Assertions.assertTrue(
SavepointUtils.savepointInProgress(sessionJob.getStatus().getJobStatus()));
Assertions.assertEquals("trigger_0", savepointInfo.getTriggerId());
- observer.observe(sessionJob, readyContext); // pending
- observer.observe(sessionJob, readyContext); // error
- flinkService.triggerSavepoint(jobID, savepointInfo, new
Configuration());
+ flinkService.triggerSavepoint(
+ jobID, SavepointTriggerType.MANUAL, savepointInfo, new
Configuration());
Assertions.assertTrue(
SavepointUtils.savepointInProgress(sessionJob.getStatus().getJobStatus()));
Assertions.assertEquals("trigger_1", savepointInfo.getTriggerId());
- flinkService.triggerSavepoint(jobID, savepointInfo, new
Configuration());
+ flinkService.triggerSavepoint(
+ jobID, SavepointTriggerType.MANUAL, savepointInfo, new
Configuration());
Assertions.assertTrue(
SavepointUtils.savepointInProgress(sessionJob.getStatus().getJobStatus()));
+ observer.observe(sessionJob, readyContext); // pending
observer.observe(sessionJob, readyContext); // success
Assertions.assertEquals("savepoint_0",
savepointInfo.getLastSavepoint().getLocation());
Assertions.assertFalse(
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 7670a34..f631141 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
@@ -34,6 +34,7 @@ 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.JobStatus;
import org.apache.flink.kubernetes.operator.crd.status.Savepoint;
+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;
import org.apache.flink.kubernetes.operator.utils.SavepointUtils;
@@ -116,6 +117,14 @@ public class ApplicationReconcilerTest {
runningJobs = flinkService.listJobs();
assertEquals(1, flinkService.getRunningCount());
assertEquals("savepoint_0", runningJobs.get(0).f0);
+ assertEquals(
+ SavepointTriggerType.UPGRADE,
+ statefulUpgrade
+ .getStatus()
+ .getJobStatus()
+ .getSavepointInfo()
+ .getLastSavepoint()
+ .getTriggerType());
deployment.getSpec().getJob().setUpgradeMode(UpgradeMode.LAST_STATE);
deployment.getSpec().setRestartNonce(100L);
@@ -146,7 +155,7 @@ public class ApplicationReconcilerTest {
.getStatus()
.getJobStatus()
.getSavepointInfo()
- .setLastSavepoint(Savepoint.of("finished_sp"));
+ .setLastSavepoint(Savepoint.of("finished_sp",
SavepointTriggerType.UPGRADE));
deployment.getStatus().getJobStatus().setState("FINISHED");
deployment.getStatus().setJobManagerDeploymentStatus(JobManagerDeploymentStatus.READY);
reconciler.reconcile(deployment, context);
@@ -236,6 +245,9 @@ public class ApplicationReconcilerTest {
assertEquals(
"trigger_0",
spDeployment.getStatus().getJobStatus().getSavepointInfo().getTriggerId());
+ assertEquals(
+ SavepointTriggerType.MANUAL,
+
spDeployment.getStatus().getJobStatus().getSavepointInfo().getTriggerType());
spDeployment.getStatus().getJobStatus().getSavepointInfo().resetTrigger();
@@ -254,6 +266,9 @@ public class ApplicationReconcilerTest {
assertEquals(
"trigger_1",
spDeployment.getStatus().getJobStatus().getSavepointInfo().getTriggerId());
+ assertEquals(
+ SavepointTriggerType.MANUAL,
+
spDeployment.getStatus().getJobStatus().getSavepointInfo().getTriggerType());
spDeployment.getStatus().getJobStatus().getSavepointInfo().resetTrigger();
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 4aff486..fd96602 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
@@ -27,6 +27,7 @@ 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.crd.status.SavepointTriggerType;
import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
import org.apache.flink.kubernetes.operator.utils.SavepointUtils;
@@ -194,6 +195,14 @@ public class FlinkSessionJobReconcilerTest {
.getSavepointInfo()
.getSavepointHistory()
.size());
+ assertEquals(
+ SavepointTriggerType.UPGRADE,
+ statefulSessionJob
+ .getStatus()
+ .getJobStatus()
+ .getSavepointInfo()
+ .getLastSavepoint()
+ .getTriggerType());
// upgraded
reconciler.reconcile(statefulSessionJob, readyContext);
@@ -299,6 +308,9 @@ public class FlinkSessionJobReconcilerTest {
assertEquals(
"trigger_0",
sp1SessionJob.getStatus().getJobStatus().getSavepointInfo().getTriggerId());
+ assertEquals(
+ SavepointTriggerType.MANUAL,
+
sp1SessionJob.getStatus().getJobStatus().getSavepointInfo().getTriggerType());
// parallelism not changed
assertEquals(
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/FlinkServiceTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/FlinkServiceTest.java
index cd6d89e..c5c823d 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/FlinkServiceTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/FlinkServiceTest.java
@@ -32,6 +32,7 @@ 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.JobManagerDeploymentStatus;
import org.apache.flink.kubernetes.operator.crd.status.JobStatus;
+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;
import org.apache.flink.runtime.messages.Acknowledge;
@@ -203,6 +204,7 @@ public class FlinkServiceTest {
flinkDeployment.getStatus().setJobStatus(jobStatus);
flinkService.triggerSavepoint(
flinkDeployment.getStatus().getJobStatus().getJobId(),
+ SavepointTriggerType.MANUAL,
flinkDeployment.getStatus().getJobStatus().getSavepointInfo(),
configuration);
assertTrue(triggerSavepointFuture.isDone());
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/SavepointUtilsTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/SavepointUtilsTest.java
new file mode 100644
index 0000000..ee004a0
--- /dev/null
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/SavepointUtilsTest.java
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.operator.utils;
+
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.kubernetes.operator.TestUtils;
+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.status.JobManagerDeploymentStatus;
+import org.apache.flink.kubernetes.operator.crd.status.SavepointTriggerType;
+
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Optional;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/** Tests for {@link SavepointUtils}. */
+public class SavepointUtilsTest {
+
+ private final FlinkConfigManager configManager = new
FlinkConfigManager(new Configuration());
+
+ @Test
+ public void testTriggering() {
+ FlinkDeployment deployment =
TestUtils.buildApplicationCluster(FlinkVersion.v1_15);
+ deployment
+ .getMetadata()
+
.setCreationTimestamp(Instant.now().minus(Duration.ofMinutes(15)).toString());
+
+
deployment.getStatus().getJobStatus().setState(JobStatus.RUNNING.name());
+
deployment.getStatus().setJobManagerDeploymentStatus(JobManagerDeploymentStatus.READY);
+
+ deployment
+ .getStatus()
+ .getReconciliationStatus()
+ .serializeAndSetLastReconciledSpec(deployment.getSpec());
+
+ assertEquals(
+ Optional.empty(),
+ SavepointUtils.shouldTriggerSavepoint(
+ deployment,
configManager.getObserveConfig(deployment)));
+
+ deployment
+ .getSpec()
+ .getFlinkConfiguration()
+
.put(KubernetesOperatorConfigOptions.PERIODIC_SAVEPOINT_INTERVAL.key(), "10m");
+ deployment
+ .getStatus()
+ .getReconciliationStatus()
+ .serializeAndSetLastReconciledSpec(deployment.getSpec());
+
+ assertEquals(
+ Optional.of(SavepointTriggerType.PERIODIC),
+ SavepointUtils.shouldTriggerSavepoint(
+ deployment,
configManager.getObserveConfig(deployment)));
+
deployment.getStatus().getJobStatus().getSavepointInfo().resetTrigger();
+
+ deployment.getSpec().getJob().setSavepointTriggerNonce(123L);
+ assertEquals(
+ Optional.of(SavepointTriggerType.MANUAL),
+ SavepointUtils.shouldTriggerSavepoint(
+ deployment,
configManager.getObserveConfig(deployment)));
+ }
+}
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/validation/DefaultValidatorTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/validation/DefaultValidatorTest.java
index 79cb46e..312eaab 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/validation/DefaultValidatorTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/validation/DefaultValidatorTest.java
@@ -38,6 +38,7 @@ 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.JobStatus;
import org.apache.flink.kubernetes.operator.crd.status.Savepoint;
+import org.apache.flink.kubernetes.operator.crd.status.SavepointTriggerType;
import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
import org.apache.flink.kubernetes.utils.Constants;
@@ -133,6 +134,20 @@ public class DefaultValidatorTest {
"Savepoint could not be manually triggered for the
running job while config key[%s] is not set",
CheckpointingOptions.SAVEPOINT_DIRECTORY.key()));
+ testError(
+ dep -> {
+ dep.getSpec()
+ .setFlinkConfiguration(
+ Map.of(
+ KubernetesOperatorConfigOptions
+
.PERIODIC_SAVEPOINT_INTERVAL
+ .key(),
+ "1m"));
+ },
+ String.format(
+ "Periodic savepoints cannot be enabled when config
key[%s] is not set",
+ CheckpointingOptions.SAVEPOINT_DIRECTORY.key()));
+
// Test conf validation
testSuccess(
dep ->
@@ -214,7 +229,7 @@ public class DefaultValidatorTest {
dep.getStatus()
.getJobStatus()
.getSavepointInfo()
- .setLastSavepoint(Savepoint.of("sp"));
+ .setLastSavepoint(Savepoint.of("sp",
SavepointTriggerType.UPGRADE));
dep.getStatus()
.setReconciliationStatus(new
FlinkDeploymentReconciliationStatus());
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 05ca231..6246a36 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
@@ -9132,11 +9132,25 @@ spec:
type: integer
location:
type: string
+ triggerType:
+ enum:
+ - MANUAL
+ - PERIODIC
+ - UPGRADE
+ - UNKNOWN
+ type: string
type: object
triggerId:
type: string
triggerTimestamp:
type: integer
+ triggerType:
+ enum:
+ - MANUAL
+ - PERIODIC
+ - UPGRADE
+ - UNKNOWN
+ type: string
savepointHistory:
items:
properties:
@@ -9144,8 +9158,17 @@ spec:
type: integer
location:
type: string
+ triggerType:
+ enum:
+ - MANUAL
+ - PERIODIC
+ - UPGRADE
+ - UNKNOWN
+ type: string
type: object
type: array
+ lastPeriodicSavepointTimestamp:
+ type: integer
type: object
type: object
error:
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 278b790..26c1fa9 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
@@ -96,11 +96,25 @@ spec:
type: integer
location:
type: string
+ triggerType:
+ enum:
+ - MANUAL
+ - PERIODIC
+ - UPGRADE
+ - UNKNOWN
+ type: string
type: object
triggerId:
type: string
triggerTimestamp:
type: integer
+ triggerType:
+ enum:
+ - MANUAL
+ - PERIODIC
+ - UPGRADE
+ - UNKNOWN
+ type: string
savepointHistory:
items:
properties:
@@ -108,8 +122,17 @@ spec:
type: integer
location:
type: string
+ triggerType:
+ enum:
+ - MANUAL
+ - PERIODIC
+ - UPGRADE
+ - UNKNOWN
+ type: string
type: object
type: array
+ lastPeriodicSavepointTimestamp:
+ type: integer
type: object
type: object
error:
