From e9061cbb59840c5d6328b64fe970672a526a0710 Mon Sep 17 00:00:00 2001
From: zhangyuhang <2827528315@qq.com>
Date: Wed, 27 Nov 2024 21:04:54 +0800
Subject: [PATCH] [Optimization][dinky-getaway] Delete the previously failed
 cluster when resubmitting the task. (#3969)

Co-authored-by: yuhang2.zhang <yuhang2.zhang@ly.com>
---
 .../apache/flink/kubernetes/KubernetesClusterDescriptor.java   | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/dinky-client/dinky-client-1.18/src/main/java/org/apache/flink/kubernetes/KubernetesClusterDescriptor.java b/dinky-client/dinky-client-1.18/src/main/java/org/apache/flink/kubernetes/KubernetesClusterDescriptor.java
index 41b782176a..27ed7870ae 100644
--- a/dinky-client/dinky-client-1.18/src/main/java/org/apache/flink/kubernetes/KubernetesClusterDescriptor.java
+++ b/dinky-client/dinky-client-1.18/src/main/java/org/apache/flink/kubernetes/KubernetesClusterDescriptor.java
@@ -171,7 +171,8 @@ public ClusterClientProvider<String> deployApplicationCluster(
             throws ClusterDeploymentException {
         if (client.getService(ExternalServiceDecorator.getExternalServiceName(clusterId))
                 .isPresent()) {
-            throw new ClusterDeploymentException("The Flink cluster " + clusterId + " already exists.");
+            client.stopAndCleanupCluster(clusterId);
+            LOG.warn("The Flink cluster {} already exists, automatically stopAndCleanupCluster.", clusterId);
         }
 
         checkNotNull(clusterSpecification);