[K8s] Fix failed application cancellation state race

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);
 

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

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());
+        }
     }
 }

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">

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;
+    }
 }

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),

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))
+  }
+
+}