[Improve] Graceful stop flink job with savepoint (#3796)

Co-authored-by: benjobs <[email protected]>
apache · Jun 26, 2024 · 45f169a · 45f169a
1 parent 8b5b200
commit 45f169a
Show file tree

Hide file tree

Showing 3 changed files with 3 additions and 5 deletions.
diff --git a/...g/apache/streampark/console/core/service/application/impl/ApplicationInfoServiceImpl.java b/...g/apache/streampark/console/core/service/application/impl/ApplicationInfoServiceImpl.java
@@ -245,7 +245,8 @@ public boolean checkEnv(Application appParam) throws ApplicationException {
   @Override
   public boolean checkAlter(Application appParam) {
     Long appId = appParam.getId();
-    if (FlinkAppStateEnum.CANCELED != appParam.getStateEnum()) {
+    if (FlinkAppStateEnum.CANCELED != appParam.getStateEnum()
+        && FlinkAppStateEnum.FINISHED != appParam.getStateEnum()) {
       return false;
     }
     long cancelUserId = FlinkAppHttpWatcher.getCanceledJobUserId(appId);

diff --git a/...service/src/main/java/org/apache/streampark/console/core/watcher/FlinkAppHttpWatcher.java b/...service/src/main/java/org/apache/streampark/console/core/watcher/FlinkAppHttpWatcher.java
@@ -483,6 +483,7 @@ private void handleNotRunState(
         doPersistMetrics(application, false);
         break;
       case CANCELED:
+      case FINISHED:
         log.info(
             "[StreamPark][FlinkAppHttpWatcher] getFromFlinkRestApi, job state {}, stop tracking and delete stopFrom!",
             currentState.name());

diff --git a/...lient-core/src/main/scala/org/apache/streampark/flink/client/trait/FlinkClientTrait.scala b/...lient-core/src/main/scala/org/apache/streampark/flink/client/trait/FlinkClientTrait.scala
@@ -538,10 +538,6 @@ trait FlinkClientTrait extends Logger {
       case (false, false) =>
         client.cancel(jobID).get()
         null
-      case (true, false) =>
-        clientWrapper
-          .cancelWithSavepoint(jobID, savePointDir, cancelRequest.nativeFormat)
-          .get()
       case (_, _) =>
         clientWrapper
           .stopWithSavepoint(