[flink-kubernetes-operator] branch release-1.0 updated: [FLINK-27889] Fix cleanup when lastReconciledSpec is null

[gyfora]

This is an automated email from the ASF dual-hosted git repository.

gyfora pushed a commit to branch release-1.0
in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git


The following commit(s) were added to refs/heads/release-1.0 by this push:
     new 6ce1c6b  [FLINK-27889] Fix cleanup when lastReconciledSpec is null
6ce1c6b is described below

commit 6ce1c6b51241ca2b866816232acb1612e2b18c40
Author: Hector Miuler Malpica Gallegos <miu...@gmail.com>
AuthorDate: Fri Jun 3 21:06:11 2022 -0500

    [FLINK-27889] Fix cleanup when lastReconciledSpec is null
---
 .../operator/reconciler/deployment/ApplicationReconciler.java      | 7 ++++++-
 .../operator/controller/FlinkDeploymentControllerTest.java         | 7 +++++++
 2 files changed, 13 insertions(+), 1 deletion(-)

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..f5beddb 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
@@ -324,7 +324,12 @@ public class ApplicationReconciler extends 
AbstractDeploymentReconciler {
     @Override
     @SneakyThrows
     protected void shutdown(FlinkDeployment flinkApp) {
-        flinkService.cancelJob(flinkApp, UpgradeMode.STATELESS);
+        var status = flinkApp.getStatus();
+        if (status.getReconciliationStatus().getLastReconciledSpec() == null) {
+            flinkService.deleteClusterDeployment(flinkApp.getMetadata(), 
status, true);
+        } else {
+            flinkService.cancelJob(flinkApp, UpgradeMode.STATELESS);
+        }
     }
 
     private void triggerSavepoint(FlinkDeployment deployment) throws Exception 
{
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 98c4337..323fcfe 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
@@ -702,6 +702,13 @@ public class FlinkDeploymentControllerTest {
                 
appCluster.getStatus().getReconciliationStatus().getLastStableSpec());
     }
 
+    @Test
+    public void cleanUpNewDeployment() {
+        FlinkDeployment flinkDeployment = TestUtils.buildApplicationCluster();
+        var deleteControl = testController.cleanup(flinkDeployment, context);
+        assertNotNull(deleteControl);
+    }
+
     private static Stream<Arguments> applicationTestParams() {
         List<Arguments> args = new ArrayList<>();
         for (FlinkVersion version : FlinkVersion.values()) {