From de2697c9e5bc350e7bdb466818ff6457f4636b1f Mon Sep 17 00:00:00 2001
From: Power John <johntomcat7408@gmail.com>
Date: Sat, 25 Jul 2026 23:57:07 +0800
Subject: [PATCH] [K8s] Fix failed application cancellation state race

Allow failed Kubernetes applications to be canceled while protecting active and completed cancellation from stale watcher snapshots. Keep the database guard scoped so interrupted cancellation recovers after Console restart, explicit restarts and terminal corrections remain authoritative, and standalone savepoint completion remains unaffected.

Refs #4332
---
 .../core/mapper/FlinkApplicationMapper.java   |   7 +-
 .../FlinkApplicationManageServiceImpl.java    |   7 +-
 .../watcher/FlinkK8sChangeEventListener.java  |   8 +-
 .../mapper/core/FlinkApplicationMapper.xml    |  37 +++-
 .../FlinkApplicationManageServiceTest.java    | 196 ++++++++++++++++++
 .../flink/app/hooks/useAppTableAction.ts      |   3 +-
 .../FlinkJobStatusWatcherTest.scala           |  54 +++++
 7 files changed, 304 insertions(+), 8 deletions(-)
 create mode 100644 streampark-flink/streampark-flink-kubernetes/src/test/scala/org/apache/streampark/flink/kubernetes/FlinkJobStatusWatcherTest.scala

diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/mapper/FlinkApplicationMapper.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/mapper/FlinkApplicationMapper.java
index 3338b090ca..5b5afc3bb5 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/mapper/FlinkApplicationMapper.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/mapper/FlinkApplicationMapper.java
@@ -33,7 +33,12 @@ public interface FlinkApplicationMapper extends BaseMapper<FlinkApplication> {
 
     FlinkApplication selectApp(@Param("id") Long id);
 
-    void persistMetrics(@Param("app") FlinkApplication application);
+    void persistMetrics(
+                        @Param("app") FlinkApplication application,
+                        @Param("cancellingState") int cancellingState,
+                        @Param("canceledState") int canceledState,
+                        @Param("cancellingOptionState") int cancellingOptionState,
+                        @Param("savepointingOptionState") int savepointingOptionState);
 
     List<FlinkApplication> selectAppsByTeamId(@Param("teamId") Long teamId);
 
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/FlinkApplicationManageServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/FlinkApplicationManageServiceImpl.java
index 0d18aab3d2..545d3514f7 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/FlinkApplicationManageServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/FlinkApplicationManageServiceImpl.java
@@ -171,7 +171,12 @@ public void toEffective(FlinkApplication appParam) {
 
     @Override
     public void persistMetrics(FlinkApplication appParam) {
-        this.baseMapper.persistMetrics(appParam);
+        this.baseMapper.persistMetrics(
+            appParam,
+            FlinkAppStateEnum.CANCELLING.getValue(),
+            FlinkAppStateEnum.CANCELED.getValue(),
+            OptionStateEnum.CANCELLING.getValue(),
+            OptionStateEnum.SAVEPOINTING.getValue());
     }
 
     @Override
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/watcher/FlinkK8sChangeEventListener.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/watcher/FlinkK8sChangeEventListener.java
index 8a57a07a17..f9379aefa8 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/watcher/FlinkK8sChangeEventListener.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/watcher/FlinkK8sChangeEventListener.java
@@ -188,8 +188,10 @@ private void setByJobStatusCV(FlinkApplication app, JobStatusCV jobStatus) {
         app.setStartTime(new Date(startTime > 0 ? startTime : 0));
         app.setEndTime(endTime > 0 && endTime >= startTime ? new Date(endTime) : null);
         app.setDuration(duration > 0 ? duration : 0);
-        // when a flink job status change event can be received, it means
-        // that the operation command sent by streampark has been completed.
-        app.setOptionState(OptionStateEnum.NONE.getValue());
+        // A non-terminal event can race with an in-flight cancellation. Keep the operation state
+        // until the watcher observes a terminal job state.
+        if (state != FlinkJobState.CANCELLING()) {
+            app.setOptionState(OptionStateEnum.NONE.getValue());
+        }
     }
 }
diff --git a/streampark-console/streampark-console-service/src/main/resources/mapper/core/FlinkApplicationMapper.xml b/streampark-console/streampark-console-service/src/main/resources/mapper/core/FlinkApplicationMapper.xml
index 0c567e6d0d..103d5a0195 100644
--- a/streampark-console/streampark-console-service/src/main/resources/mapper/core/FlinkApplicationMapper.xml
+++ b/streampark-console/streampark-console-service/src/main/resources/mapper/core/FlinkApplicationMapper.xml
@@ -124,7 +124,22 @@
                 tracking=#{app.tracking},
             </if>
             <if test="app.optionState != null">
-                option_state=#{app.optionState},
+                <choose>
+                    <when test="@org.apache.streampark.console.core.enums.FlinkAppStateEnum@isEndState(app.state)">
+                        option_state=#{app.optionState},
+                    </when>
+                    <otherwise>
+                        option_state=case
+                            when state=#{cancellingState}
+                                and option_state in (
+                                    #{cancellingOptionState},
+                                    #{savepointingOptionState}
+                                )
+                            then option_state
+                            else #{app.optionState}
+                        end,
+                    </otherwise>
+                </choose>
             </if>
             <if test="app.startTime != null">
                 start_time=#{app.startTime},
@@ -165,9 +180,27 @@
                     </if>
                 </otherwise>
             </choose>
-            state=#{app.state}
+            <choose>
+                <when test="@org.apache.streampark.console.core.enums.FlinkAppStateEnum@isEndState(app.state)">
+                    state=#{app.state},
+                </when>
+                <otherwise>
+                    state=case
+                        when state=#{cancellingState}
+                            and option_state in (
+                                #{cancellingOptionState},
+                                #{savepointingOptionState}
+                            )
+                        then state
+                        else #{app.state}
+                    end,
+                </otherwise>
+            </choose>
         </set>
         where id=#{app.id}
+        <if test="@org.apache.streampark.console.core.enums.FlinkAppStateEnum@isEndState(app.state) == false">
+            and state != #{canceledState}
+        </if>
     </update>
 
     <select id="selectAppsByTeamId" resultType="org.apache.streampark.console.core.entity.FlinkApplication" parameterType="java.lang.Long">
diff --git a/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/core/service/FlinkApplicationManageServiceTest.java b/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/core/service/FlinkApplicationManageServiceTest.java
index 5724ecee9f..f8f4c3181e 100644
--- a/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/core/service/FlinkApplicationManageServiceTest.java
+++ b/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/core/service/FlinkApplicationManageServiceTest.java
@@ -21,6 +21,8 @@
 import org.apache.streampark.console.SpringUnitTestBase;
 import org.apache.streampark.console.core.entity.FlinkApplication;
 import org.apache.streampark.console.core.entity.YarnQueue;
+import org.apache.streampark.console.core.enums.FlinkAppStateEnum;
+import org.apache.streampark.console.core.enums.OptionStateEnum;
 import org.apache.streampark.console.core.service.application.FlinkApplicationActionService;
 import org.apache.streampark.console.core.service.application.FlinkApplicationManageService;
 import org.apache.streampark.console.core.service.application.impl.FlinkApplicationManageServiceImpl;
@@ -146,4 +148,198 @@ void testCheckQueueValidationIfNeeded() {
         app2.setYarnQueue(nonExistedQueue);
         assertThat(applicationServiceImpl.validateQueueIfNeeded(app1, app2)).isFalse();
     }
+
+    @Test
+    void testPersistMetricsPreservesInFlightCancellation() {
+        assertNonTerminalMetricsPreserveOperation(OptionStateEnum.CANCELLING);
+        assertNonTerminalMetricsPreserveOperation(OptionStateEnum.SAVEPOINTING);
+    }
+
+    @Test
+    void testPersistMetricsRecoversInterruptedCancellation() {
+        assertInterruptedCancellationRecovers(FlinkAppStateEnum.RUNNING);
+        assertInterruptedCancellationRecovers(FlinkAppStateEnum.FAILING);
+    }
+
+    @Test
+    void testPersistMetricsUpdatesUnprotectedNonTerminalState() {
+        FlinkApplication persisted = createApplication(
+            FlinkAppStateEnum.RUNNING, OptionStateEnum.NONE);
+
+        FlinkApplication snapshot = new FlinkApplication();
+        snapshot.setId(persisted.getId());
+        snapshot.setState(FlinkAppStateEnum.FAILING.getValue());
+        snapshot.setOptionState(OptionStateEnum.NONE.getValue());
+        applicationManageService.persistMetrics(snapshot);
+
+        FlinkApplication actual = applicationManageService.getById(persisted.getId());
+        assertThat(actual.getState()).isEqualTo(FlinkAppStateEnum.FAILING.getValue());
+        assertThat(actual.getOptionState()).isEqualTo(OptionStateEnum.NONE.getValue());
+    }
+
+    @Test
+    void testPersistMetricsDoesNotBlockStandaloneSavepointCompletion() {
+        FlinkApplication persisted = createApplication(
+            FlinkAppStateEnum.RUNNING, OptionStateEnum.SAVEPOINTING);
+
+        FlinkApplication completedSnapshot = new FlinkApplication();
+        completedSnapshot.setId(persisted.getId());
+        completedSnapshot.setState(FlinkAppStateEnum.RUNNING.getValue());
+        completedSnapshot.setOptionState(OptionStateEnum.NONE.getValue());
+        applicationManageService.persistMetrics(completedSnapshot);
+
+        FlinkApplication actual = applicationManageService.getById(persisted.getId());
+        assertThat(actual.getState()).isEqualTo(FlinkAppStateEnum.RUNNING.getValue());
+        assertThat(actual.getOptionState()).isEqualTo(OptionStateEnum.NONE.getValue());
+    }
+
+    @Test
+    void testPersistMetricsConvergesCancellationOnTerminalState() {
+        FlinkApplication persisted = createApplication(
+            FlinkAppStateEnum.CANCELLING, OptionStateEnum.CANCELLING);
+
+        FlinkApplication terminalSnapshot = new FlinkApplication();
+        terminalSnapshot.setId(persisted.getId());
+        terminalSnapshot.setState(FlinkAppStateEnum.CANCELED.getValue());
+        terminalSnapshot.setOptionState(OptionStateEnum.NONE.getValue());
+        applicationManageService.persistMetrics(terminalSnapshot);
+
+        FlinkApplication actual = applicationManageService.getById(persisted.getId());
+        assertThat(actual.getState()).isEqualTo(FlinkAppStateEnum.CANCELED.getValue());
+        assertThat(actual.getOptionState()).isEqualTo(OptionStateEnum.NONE.getValue());
+    }
+
+    @Test
+    void testPersistMetricsRejectsLateNonTerminalSnapshotAfterCancellationCompletes() {
+        assertCompletedCancellationRejectsLateSnapshot(
+            FlinkAppStateEnum.FAILING, OptionStateEnum.NONE);
+        assertCompletedCancellationRejectsLateSnapshot(
+            FlinkAppStateEnum.RUNNING, OptionStateEnum.NONE);
+        assertCompletedCancellationRejectsLateSnapshot(
+            FlinkAppStateEnum.CANCELLING, OptionStateEnum.CANCELLING);
+    }
+
+    @Test
+    void testPersistMetricsAllowsExplicitRestartAfterCancellationCompletes() {
+        FlinkApplication persisted = createApplication(
+            FlinkAppStateEnum.CANCELED, OptionStateEnum.NONE);
+
+        FlinkApplication starting = new FlinkApplication();
+        starting.setId(persisted.getId());
+        starting.setState(FlinkAppStateEnum.STARTING.getValue());
+        assertThat(applicationManageService.updateById(starting)).isTrue();
+
+        FlinkApplication runningSnapshot = new FlinkApplication();
+        runningSnapshot.setId(persisted.getId());
+        runningSnapshot.setState(FlinkAppStateEnum.RUNNING.getValue());
+        runningSnapshot.setOptionState(OptionStateEnum.NONE.getValue());
+        runningSnapshot.setTotalTask(2);
+        applicationManageService.persistMetrics(runningSnapshot);
+
+        FlinkApplication actual = applicationManageService.getById(persisted.getId());
+        assertThat(actual.getState()).isEqualTo(FlinkAppStateEnum.RUNNING.getValue());
+        assertThat(actual.getOptionState()).isEqualTo(OptionStateEnum.NONE.getValue());
+        assertThat(actual.getTotalTask()).isEqualTo(2);
+    }
+
+    @Test
+    void testPersistMetricsAllowsTerminalCorrectionAfterCancellationCompletes() {
+        FlinkApplication persisted = createApplication(
+            FlinkAppStateEnum.CANCELED, OptionStateEnum.NONE);
+
+        FlinkApplication failedSnapshot = new FlinkApplication();
+        failedSnapshot.setId(persisted.getId());
+        failedSnapshot.setState(FlinkAppStateEnum.FAILED.getValue());
+        failedSnapshot.setOptionState(OptionStateEnum.NONE.getValue());
+        applicationManageService.persistMetrics(failedSnapshot);
+
+        FlinkApplication actual = applicationManageService.getById(persisted.getId());
+        assertThat(actual.getState()).isEqualTo(FlinkAppStateEnum.FAILED.getValue());
+        assertThat(actual.getOptionState()).isEqualTo(OptionStateEnum.NONE.getValue());
+    }
+
+    @Test
+    void testPersistMetricsDoesNotFreezeRecoverableLostState() {
+        FlinkApplication persisted = createApplication(
+            FlinkAppStateEnum.LOST, OptionStateEnum.NONE);
+
+        FlinkApplication recoveredSnapshot = new FlinkApplication();
+        recoveredSnapshot.setId(persisted.getId());
+        recoveredSnapshot.setState(FlinkAppStateEnum.RUNNING.getValue());
+        recoveredSnapshot.setOptionState(OptionStateEnum.NONE.getValue());
+        applicationManageService.persistMetrics(recoveredSnapshot);
+
+        FlinkApplication actual = applicationManageService.getById(persisted.getId());
+        assertThat(actual.getState()).isEqualTo(FlinkAppStateEnum.RUNNING.getValue());
+        assertThat(actual.getOptionState()).isEqualTo(OptionStateEnum.NONE.getValue());
+    }
+
+    private void assertCompletedCancellationRejectsLateSnapshot(
+                                                                FlinkAppStateEnum lateState,
+                                                                OptionStateEnum lateOptionState) {
+        FlinkApplication persisted = createApplication(
+            FlinkAppStateEnum.CANCELLING, OptionStateEnum.CANCELLING);
+
+        FlinkApplication terminalSnapshot = new FlinkApplication();
+        terminalSnapshot.setId(persisted.getId());
+        terminalSnapshot.setState(FlinkAppStateEnum.CANCELED.getValue());
+        terminalSnapshot.setOptionState(OptionStateEnum.NONE.getValue());
+        applicationManageService.persistMetrics(terminalSnapshot);
+
+        FlinkApplication lateSnapshot = new FlinkApplication();
+        lateSnapshot.setId(persisted.getId());
+        lateSnapshot.setState(lateState.getValue());
+        lateSnapshot.setOptionState(lateOptionState.getValue());
+        lateSnapshot.setTotalTask(7);
+        applicationManageService.persistMetrics(lateSnapshot);
+
+        FlinkApplication actual = applicationManageService.getById(persisted.getId());
+        assertThat(actual.getState()).isEqualTo(FlinkAppStateEnum.CANCELED.getValue());
+        assertThat(actual.getOptionState()).isEqualTo(OptionStateEnum.NONE.getValue());
+        assertThat(actual.getTotalTask()).isNull();
+    }
+
+    private void assertInterruptedCancellationRecovers(FlinkAppStateEnum currentState) {
+        FlinkApplication persisted = createApplication(
+            FlinkAppStateEnum.CANCELLING, OptionStateEnum.NONE);
+
+        FlinkApplication currentSnapshot = new FlinkApplication();
+        currentSnapshot.setId(persisted.getId());
+        currentSnapshot.setState(currentState.getValue());
+        currentSnapshot.setOptionState(OptionStateEnum.NONE.getValue());
+        currentSnapshot.setTotalTask(4);
+        applicationManageService.persistMetrics(currentSnapshot);
+
+        FlinkApplication actual = applicationManageService.getById(persisted.getId());
+        assertThat(actual.getState()).isEqualTo(currentState.getValue());
+        assertThat(actual.getOptionState()).isEqualTo(OptionStateEnum.NONE.getValue());
+        assertThat(actual.getTotalTask()).isEqualTo(4);
+    }
+
+    private void assertNonTerminalMetricsPreserveOperation(OptionStateEnum optionState) {
+        FlinkApplication persisted = createApplication(FlinkAppStateEnum.CANCELLING, optionState);
+
+        FlinkApplication staleSnapshot = new FlinkApplication();
+        staleSnapshot.setId(persisted.getId());
+        staleSnapshot.setState(FlinkAppStateEnum.FAILING.getValue());
+        staleSnapshot.setOptionState(OptionStateEnum.NONE.getValue());
+        staleSnapshot.setTotalTask(3);
+        applicationManageService.persistMetrics(staleSnapshot);
+
+        FlinkApplication actual = applicationManageService.getById(persisted.getId());
+        assertThat(actual.getState()).isEqualTo(FlinkAppStateEnum.CANCELLING.getValue());
+        assertThat(actual.getOptionState()).isEqualTo(optionState.getValue());
+        assertThat(actual.getTotalTask()).isEqualTo(3);
+    }
+
+    private FlinkApplication createApplication(
+                                               FlinkAppStateEnum state,
+                                               OptionStateEnum optionState) {
+        FlinkApplication application = new FlinkApplication();
+        application.setTeamId(1L);
+        application.setState(state.getValue());
+        application.setOptionState(optionState.getValue());
+        assertThat(applicationManageService.save(application)).isTrue();
+        return application;
+    }
 }
diff --git a/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useAppTableAction.ts b/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useAppTableAction.ts
index 167b63fde7..91be490b20 100644
--- a/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useAppTableAction.ts
+++ b/streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useAppTableAction.ts
@@ -104,7 +104,8 @@ export const useAppTableAction = (
         class: 'e2e-flinkapp-cancel-btn',
         tooltip: { title: t('flink.app.operation.cancel') },
         ifShow:
-          record.state == AppStateEnum.RUNNING && record['optionState'] == OptionStateEnum.NONE,
+          [AppStateEnum.RUNNING, AppStateEnum.FAILING].includes(record.state) &&
+          record['optionState'] == OptionStateEnum.NONE,
         auth: 'app:cancel',
         icon: 'ant-design:pause-circle-outlined',
         onClick: handleCancel.bind(null, record),
diff --git a/streampark-flink/streampark-flink-kubernetes/src/test/scala/org/apache/streampark/flink/kubernetes/FlinkJobStatusWatcherTest.scala b/streampark-flink/streampark-flink-kubernetes/src/test/scala/org/apache/streampark/flink/kubernetes/FlinkJobStatusWatcherTest.scala
new file mode 100644
index 0000000000..c942be2d81
--- /dev/null
+++ b/streampark-flink/streampark-flink-kubernetes/src/test/scala/org/apache/streampark/flink/kubernetes/FlinkJobStatusWatcherTest.scala
@@ -0,0 +1,54 @@
+/*
+ * 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.streampark.flink.kubernetes
+
+import org.apache.streampark.flink.kubernetes.enums.FlinkJobState
+import org.apache.streampark.flink.kubernetes.watcher.FlinkJobStatusWatcher
+
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Test
+
+class FlinkJobStatusWatcherTest {
+
+  @Test
+  def doNotInferActiveCancellationFromPersistedState(): Unit = {
+    Seq(FlinkJobState.RUNNING, FlinkJobState.FAILING, FlinkJobState.SILENT).foreach {
+      current =>
+        assertEquals(
+          current,
+          FlinkJobStatusWatcher.inferFlinkJobStateFromPersist(
+            current,
+            FlinkJobState.CANCELLING))
+    }
+  }
+
+  @Test
+  def convergeCancellationWhenWatcherObservesTerminalState(): Unit = {
+    assertEquals(
+      FlinkJobState.CANCELED,
+      FlinkJobStatusWatcher.inferFlinkJobStateFromPersist(
+        FlinkJobState.TERMINATED,
+        FlinkJobState.CANCELLING))
+    assertEquals(
+      FlinkJobState.FAILED,
+      FlinkJobStatusWatcher.inferFlinkJobStateFromPersist(
+        FlinkJobState.FAILED,
+        FlinkJobState.CANCELLING))
+  }
+
+}