Skip to content

Commit 9cad469

Browse files
johntomcat7408-cmykjohntomcat7408-cmyk
committed
[K8s] Fix failed application cancellation state race
Allow failed Kubernetes applications to be canceled while keeping an in-flight cancellation stable against stale watcher updates. Terminal states still converge and standalone savepoint completion remains unaffected. Refs #4332
1 parent 0a59e04 commit 9cad469

8 files changed

Lines changed: 176 additions & 8 deletions

File tree

‎streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/mapper/FlinkApplicationMapper.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,9 @@ public interface FlinkApplicationMapper extends BaseMapper<FlinkApplication> {
3333

3434
FlinkApplication selectApp(@Param("id") Long id);
3535

36-
void persistMetrics(@Param("app") FlinkApplication application);
36+
void persistMetrics(
37+
@Param("app") FlinkApplication application,
38+
@Param("cancellingState") int cancellingState);
3739

3840
List<FlinkApplication> selectAppsByTeamId(@Param("teamId") Long teamId);
3941

‎streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/FlinkApplicationManageServiceImpl.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -171,7 +171,9 @@ public void toEffective(FlinkApplication appParam) {
171171

172172
@Override
173173
public void persistMetrics(FlinkApplication appParam) {
174-
this.baseMapper.persistMetrics(appParam);
174+
this.baseMapper.persistMetrics(
175+
appParam,
176+
FlinkAppStateEnum.CANCELLING.getValue());
175177
}
176178

177179
@Override

‎streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/watcher/FlinkK8sChangeEventListener.java‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -188,8 +188,10 @@ private void setByJobStatusCV(FlinkApplication app, JobStatusCV jobStatus) {
188188
app.setStartTime(new Date(startTime > 0 ? startTime : 0));
189189
app.setEndTime(endTime > 0 && endTime >= startTime ? new Date(endTime) : null);
190190
app.setDuration(duration > 0 ? duration : 0);
191-
// when a flink job status change event can be received, it means
192-
// that the operation command sent by streampark has been completed.
193-
app.setOptionState(OptionStateEnum.NONE.getValue());
191+
// A non-terminal event can race with an in-flight cancellation. Keep the operation state
192+
// until the watcher observes a terminal job state.
193+
if (state != FlinkJobState.CANCELLING()) {
194+
app.setOptionState(OptionStateEnum.NONE.getValue());
195+
}
194196
}
195197
}

‎streampark-console/streampark-console-service/src/main/resources/mapper/core/FlinkApplicationMapper.xml‎

Lines changed: 24 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -124,7 +124,18 @@
124124
tracking=#{app.tracking},
125125
</if>
126126
<if test="app.optionState != null">
127-
option_state=#{app.optionState},
127+
<choose>
128+
<when test="@org.apache.streampark.console.core.enums.FlinkAppStateEnum@isEndState(app.state)">
129+
option_state=#{app.optionState},
130+
</when>
131+
<otherwise>
132+
option_state=case
133+
when state=#{cancellingState}
134+
then option_state
135+
else #{app.optionState}
136+
end,
137+
</otherwise>
138+
</choose>
128139
</if>
129140
<if test="app.startTime != null">
130141
start_time=#{app.startTime},
@@ -165,7 +176,18 @@
165176
</if>
166177
</otherwise>
167178
</choose>
168-
state=#{app.state}
179+
<choose>
180+
<when test="@org.apache.streampark.console.core.enums.FlinkAppStateEnum@isEndState(app.state)">
181+
state=#{app.state},
182+
</when>
183+
<otherwise>
184+
state=case
185+
when state=#{cancellingState}
186+
then state
187+
else #{app.state}
188+
end,
189+
</otherwise>
190+
</choose>
169191
</set>
170192
where id=#{app.id}
171193
</update>

‎streampark-console/streampark-console-service/src/test/java/org/apache/streampark/console/core/service/FlinkApplicationManageServiceTest.java‎

Lines changed: 83 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,8 @@
2121
import org.apache.streampark.console.SpringUnitTestBase;
2222
import org.apache.streampark.console.core.entity.FlinkApplication;
2323
import org.apache.streampark.console.core.entity.YarnQueue;
24+
import org.apache.streampark.console.core.enums.FlinkAppStateEnum;
25+
import org.apache.streampark.console.core.enums.OptionStateEnum;
2426
import org.apache.streampark.console.core.service.application.FlinkApplicationActionService;
2527
import org.apache.streampark.console.core.service.application.FlinkApplicationManageService;
2628
import org.apache.streampark.console.core.service.application.impl.FlinkApplicationManageServiceImpl;
@@ -146,4 +148,85 @@ void testCheckQueueValidationIfNeeded() {
146148
app2.setYarnQueue(nonExistedQueue);
147149
assertThat(applicationServiceImpl.validateQueueIfNeeded(app1, app2)).isFalse();
148150
}
151+
152+
@Test
153+
void testPersistMetricsPreservesInFlightCancellation() {
154+
assertNonTerminalMetricsPreserveOperation(OptionStateEnum.CANCELLING);
155+
assertNonTerminalMetricsPreserveOperation(OptionStateEnum.SAVEPOINTING);
156+
}
157+
158+
@Test
159+
void testPersistMetricsUpdatesUnprotectedNonTerminalState() {
160+
FlinkApplication persisted = createApplication(
161+
FlinkAppStateEnum.RUNNING, OptionStateEnum.NONE);
162+
163+
FlinkApplication snapshot = new FlinkApplication();
164+
snapshot.setId(persisted.getId());
165+
snapshot.setState(FlinkAppStateEnum.FAILING.getValue());
166+
snapshot.setOptionState(OptionStateEnum.NONE.getValue());
167+
applicationManageService.persistMetrics(snapshot);
168+
169+
FlinkApplication actual = applicationManageService.getById(persisted.getId());
170+
assertThat(actual.getState()).isEqualTo(FlinkAppStateEnum.FAILING.getValue());
171+
assertThat(actual.getOptionState()).isEqualTo(OptionStateEnum.NONE.getValue());
172+
}
173+
174+
@Test
175+
void testPersistMetricsDoesNotBlockStandaloneSavepointCompletion() {
176+
FlinkApplication persisted = createApplication(
177+
FlinkAppStateEnum.RUNNING, OptionStateEnum.SAVEPOINTING);
178+
179+
FlinkApplication completedSnapshot = new FlinkApplication();
180+
completedSnapshot.setId(persisted.getId());
181+
completedSnapshot.setState(FlinkAppStateEnum.RUNNING.getValue());
182+
completedSnapshot.setOptionState(OptionStateEnum.NONE.getValue());
183+
applicationManageService.persistMetrics(completedSnapshot);
184+
185+
FlinkApplication actual = applicationManageService.getById(persisted.getId());
186+
assertThat(actual.getState()).isEqualTo(FlinkAppStateEnum.RUNNING.getValue());
187+
assertThat(actual.getOptionState()).isEqualTo(OptionStateEnum.NONE.getValue());
188+
}
189+
190+
@Test
191+
void testPersistMetricsConvergesCancellationOnTerminalState() {
192+
FlinkApplication persisted = createApplication(
193+
FlinkAppStateEnum.CANCELLING, OptionStateEnum.CANCELLING);
194+
195+
FlinkApplication terminalSnapshot = new FlinkApplication();
196+
terminalSnapshot.setId(persisted.getId());
197+
terminalSnapshot.setState(FlinkAppStateEnum.CANCELED.getValue());
198+
terminalSnapshot.setOptionState(OptionStateEnum.NONE.getValue());
199+
applicationManageService.persistMetrics(terminalSnapshot);
200+
201+
FlinkApplication actual = applicationManageService.getById(persisted.getId());
202+
assertThat(actual.getState()).isEqualTo(FlinkAppStateEnum.CANCELED.getValue());
203+
assertThat(actual.getOptionState()).isEqualTo(OptionStateEnum.NONE.getValue());
204+
}
205+
206+
private void assertNonTerminalMetricsPreserveOperation(OptionStateEnum optionState) {
207+
FlinkApplication persisted = createApplication(FlinkAppStateEnum.CANCELLING, optionState);
208+
209+
FlinkApplication staleSnapshot = new FlinkApplication();
210+
staleSnapshot.setId(persisted.getId());
211+
staleSnapshot.setState(FlinkAppStateEnum.FAILING.getValue());
212+
staleSnapshot.setOptionState(OptionStateEnum.NONE.getValue());
213+
staleSnapshot.setTotalTask(3);
214+
applicationManageService.persistMetrics(staleSnapshot);
215+
216+
FlinkApplication actual = applicationManageService.getById(persisted.getId());
217+
assertThat(actual.getState()).isEqualTo(FlinkAppStateEnum.CANCELLING.getValue());
218+
assertThat(actual.getOptionState()).isEqualTo(optionState.getValue());
219+
assertThat(actual.getTotalTask()).isEqualTo(3);
220+
}
221+
222+
private FlinkApplication createApplication(
223+
FlinkAppStateEnum state,
224+
OptionStateEnum optionState) {
225+
FlinkApplication application = new FlinkApplication();
226+
application.setTeamId(1L);
227+
application.setState(state.getValue());
228+
application.setOptionState(optionState.getValue());
229+
assertThat(applicationManageService.save(application)).isTrue();
230+
return application;
231+
}
149232
}

‎streampark-console/streampark-console-webapp/src/views/flink/app/hooks/useAppTableAction.ts‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -104,7 +104,8 @@ export const useAppTableAction = (
104104
class: 'e2e-flinkapp-cancel-btn',
105105
tooltip: { title: t('flink.app.operation.cancel') },
106106
ifShow:
107-
record.state == AppStateEnum.RUNNING && record['optionState'] == OptionStateEnum.NONE,
107+
[AppStateEnum.RUNNING, AppStateEnum.FAILING].includes(record.state) &&
108+
record['optionState'] == OptionStateEnum.NONE,
108109
auth: 'app:cancel',
109110
icon: 'ant-design:pause-circle-outlined',
110111
onClick: handleCancel.bind(null, record),

‎streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/watcher/FlinkJobStatusWatcher.scala‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -396,6 +396,8 @@ object FlinkJobStatusWatcher {
396396
case _ => FlinkJobState.TERMINATED
397397
}
398398
}
399+
case _ if previous == FlinkJobState.CANCELLING && !FlinkJobState.isEndState(current) =>
400+
FlinkJobState.CANCELLING
399401
case _ => current
400402
}
401403
}

‎streampark-flink/streampark-flink-kubernetes/src/test/scala/org/apache/streampark/flink/kubernetes/FlinkJobStatusWatcherTest.scala‎

Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,54 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.streampark.flink.kubernetes
19+
20+
import org.apache.streampark.flink.kubernetes.enums.FlinkJobState
21+
import org.apache.streampark.flink.kubernetes.watcher.FlinkJobStatusWatcher
22+
23+
import org.junit.jupiter.api.Assertions.assertEquals
24+
import org.junit.jupiter.api.Test
25+
26+
class FlinkJobStatusWatcherTest {
27+
28+
@Test
29+
def preserveCancellationForNonTerminalWatcherStates(): Unit = {
30+
Seq(FlinkJobState.RUNNING, FlinkJobState.FAILING, FlinkJobState.SILENT).foreach {
31+
current =>
32+
assertEquals(
33+
FlinkJobState.CANCELLING,
34+
FlinkJobStatusWatcher.inferFlinkJobStateFromPersist(
35+
current,
36+
FlinkJobState.CANCELLING))
37+
}
38+
}
39+
40+
@Test
41+
def convergeCancellationWhenWatcherObservesTerminalState(): Unit = {
42+
assertEquals(
43+
FlinkJobState.CANCELED,
44+
FlinkJobStatusWatcher.inferFlinkJobStateFromPersist(
45+
FlinkJobState.TERMINATED,
46+
FlinkJobState.CANCELLING))
47+
assertEquals(
48+
FlinkJobState.FAILED,
49+
FlinkJobStatusWatcher.inferFlinkJobStateFromPersist(
50+
FlinkJobState.FAILED,
51+
FlinkJobState.CANCELLING))
52+
}
53+
54+
}

0 commit comments

Comments
 (0)