You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gy...@apache.org on 2022/05/17 06:16:05 UTC
[flink-kubernetes-operator] 01/01: [hotfix] Add some missing savepoint history tests
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
commit ba06ce8e49d95b50dd50e9950a8e96bbe4d5c707
Author: Gyula Fora <g_...@apple.com>
AuthorDate: Tue May 17 08:10:47 2022 +0200
[hotfix] Add some missing savepoint history tests
Closes #216
---
.../operator/crd/status/SavepointInfo.java | 5 +----
.../controller/FlinkDeploymentControllerTest.java | 23 ++++++++++++++++++++++
.../operator/observer/SavepointObserverTest.java | 2 +-
.../sessionjob/FlinkSessionJobReconcilerTest.java | 15 ++++++++++++++
4 files changed, 40 insertions(+), 5 deletions(-)
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 b31f23d..2bd52eb 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
@@ -44,7 +44,7 @@ public class SavepointInfo {
private Long triggerTimestamp;
/** List of recent savepoints. */
- private List<Savepoint> savepointHistory;
+ private List<Savepoint> savepointHistory = new ArrayList<>();
public void setTrigger(String triggerId) {
this.triggerId = triggerId;
@@ -67,9 +67,6 @@ public class SavepointInfo {
* @param newSavepoint
*/
public void addSavepointToHistory(Savepoint newSavepoint) {
- if (savepointHistory == null) {
- savepointHistory = new ArrayList<>();
- }
if (!savepointHistory.isEmpty()) {
Savepoint recentSp = savepointHistory.get(savepointHistory.size() - 1);
if (recentSp.getLocation().equals(newSavepoint.getLocation())) {
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentControllerTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentControllerTest.java
index 0998214..e0bd5d4 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentControllerTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentControllerTest.java
@@ -302,6 +302,13 @@ public class FlinkDeploymentControllerTest {
// Upgrade job
appCluster.getSpec().getJob().setParallelism(100);
+ assertTrue(
+ appCluster
+ .getStatus()
+ .getJobStatus()
+ .getSavepointInfo()
+ .getSavepointHistory()
+ .isEmpty());
assertEquals(0, testController.reconcile(appCluster, context).getScheduleDelay().get());
assertEquals(
JobState.SUSPENDED,
@@ -311,6 +318,14 @@ public class FlinkDeploymentControllerTest {
.deserializeLastReconciledSpec()
.getJob()
.getState());
+ assertEquals(
+ 1,
+ appCluster
+ .getStatus()
+ .getJobStatus()
+ .getSavepointInfo()
+ .getSavepointHistory()
+ .size());
flinkService.setDeployFailure(true);
assertNotEquals(0, testController.reconcile(appCluster, context).getScheduleDelay().get());
@@ -321,6 +336,14 @@ public class FlinkDeploymentControllerTest {
assertEquals(1, jobs.size());
assertEquals("savepoint_0", jobs.get(0).f0);
testController.reconcile(appCluster, context);
+ assertEquals(
+ 1,
+ appCluster
+ .getStatus()
+ .getJobStatus()
+ .getSavepointInfo()
+ .getSavepointHistory()
+ .size());
// Suspend job
appCluster.getSpec().getJob().setState(JobState.SUSPENDED);
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 a1677ab..c5d18c4 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
@@ -42,7 +42,7 @@ public class SavepointObserverTest {
SavepointObserver observer =
new SavepointObserver(flinkService, configManager, new TestingStatusHelper<>());
SavepointInfo spInfo = new SavepointInfo();
- Assertions.assertNull(spInfo.getSavepointHistory());
+ Assertions.assertTrue(spInfo.getSavepointHistory().isEmpty());
Savepoint sp = new Savepoint(1, "sp1");
observer.updateSavepointHistory(spInfo, sp, configManager.getDefaultConfig());
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 645cf84..1aea732 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
@@ -168,6 +168,13 @@ public class FlinkSessionJobReconcilerTest {
reconciler.reconcile(sessionJob, readyContext);
// start the job
assertEquals(1, flinkService.listSessionJobs().size());
+ assertTrue(
+ sessionJob
+ .getStatus()
+ .getJobStatus()
+ .getSavepointInfo()
+ .getSavepointHistory()
+ .isEmpty());
// update job spec
var statefulSessionJob = ReconciliationUtils.clone(sessionJob);
@@ -178,6 +185,14 @@ public class FlinkSessionJobReconcilerTest {
// job suspended first
assertTrue(flinkService.listSessionJobs().isEmpty());
verifyJobState(statefulSessionJob, JobState.SUSPENDED, JobState.SUSPENDED.name());
+ assertEquals(
+ 1,
+ statefulSessionJob
+ .getStatus()
+ .getJobStatus()
+ .getSavepointInfo()
+ .getSavepointHistory()
+ .size());
// upgraded
reconciler.reconcile(statefulSessionJob, readyContext);