You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gy...@apache.org on 2026/01/14 15:22:02 UTC
(flink-kubernetes-operator) branch main updated: [FLINK-38077] Make sure jobmanager is accessible when trying to cancel for suspend/upgrade
This is an automated email from the ASF dual-hosted git repository.
gyfora pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git
The following commit(s) were added to refs/heads/main by this push:
new 824d96fa [FLINK-38077] Make sure jobmanager is accessible when trying to cancel for suspend/upgrade
824d96fa is described below
commit 824d96facaafd61d15b92ae1bf80ab8e76b80a85
Author: Gyula Fora <g_...@apple.com>
AuthorDate: Thu Jul 10 12:46:06 2025 +0200
[FLINK-38077] Make sure jobmanager is accessible when trying to cancel for suspend/upgrade
---
.../operator/api/status/CommonStatus.java | 12 ++++++
.../operator/api/status/FlinkDeploymentStatus.java | 7 ++++
.../api/status/JobManagerDeploymentStatus.java | 4 ++
.../operator/reconciler/ReconciliationUtils.java | 4 --
.../deployment/AbstractJobReconciler.java | 4 +-
.../deployment/ApplicationReconciler.java | 7 ++--
.../operator/service/AbstractFlinkService.java | 2 +-
.../ApplicationReconcilerUpgradeModeTest.java | 45 ++++++++++++++++------
8 files changed, 63 insertions(+), 22 deletions(-)
diff --git a/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/status/CommonStatus.java b/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/status/CommonStatus.java
index 30649e89..0de792da 100644
--- a/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/status/CommonStatus.java
+++ b/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/status/CommonStatus.java
@@ -22,6 +22,7 @@ import org.apache.flink.kubernetes.operator.api.lifecycle.ResourceLifecycleState
import org.apache.flink.kubernetes.operator.api.spec.AbstractFlinkSpec;
import org.apache.flink.kubernetes.operator.api.spec.JobState;
+import com.fasterxml.jackson.annotation.JsonIgnore;
import io.fabric8.crd.generator.annotation.PrinterColumn;
import lombok.AllArgsConstructor;
import lombok.Data;
@@ -29,6 +30,8 @@ import lombok.NoArgsConstructor;
import lombok.experimental.SuperBuilder;
import org.apache.commons.lang3.StringUtils;
+import static org.apache.flink.api.common.JobStatus.RECONCILING;
+
/** Last observed common status of the Flink deployment/Flink SessionJob. */
@Experimental
@Data
@@ -121,4 +124,13 @@ public abstract class CommonStatus<SPEC extends AbstractFlinkSpec> {
return ResourceLifecycleState.DEPLOYED;
}
+
+ @JsonIgnore
+ public boolean isJobCancellable() {
+ var jobState = jobStatus.getState();
+ if (jobState == null) {
+ return false;
+ }
+ return RECONCILING != jobState && !jobState.isGloballyTerminalState();
+ }
}
diff --git a/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/status/FlinkDeploymentStatus.java b/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/status/FlinkDeploymentStatus.java
index 136d3415..9a341299 100644
--- a/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/status/FlinkDeploymentStatus.java
+++ b/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/status/FlinkDeploymentStatus.java
@@ -20,6 +20,7 @@ package org.apache.flink.kubernetes.operator.api.status;
import org.apache.flink.annotation.Experimental;
import org.apache.flink.kubernetes.operator.api.spec.FlinkDeploymentSpec;
+import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import lombok.AllArgsConstructor;
import lombok.Data;
@@ -55,4 +56,10 @@ public class FlinkDeploymentStatus extends CommonStatus<FlinkDeploymentSpec> {
/** Information about the TaskManagers for the scale subresource. */
private TaskManagerInfo taskManager;
+
+ @JsonIgnore
+ @Override
+ public boolean isJobCancellable() {
+ return super.isJobCancellable() && jobManagerDeploymentStatus.isRestApiAvailable();
+ }
}
diff --git a/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/status/JobManagerDeploymentStatus.java b/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/status/JobManagerDeploymentStatus.java
index 54a0181b..2c86c49a 100644
--- a/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/status/JobManagerDeploymentStatus.java
+++ b/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/status/JobManagerDeploymentStatus.java
@@ -35,4 +35,8 @@ public enum JobManagerDeploymentStatus {
/** Deployment in terminal error, requires spec change for reconciliation to continue. */
ERROR;
+
+ public boolean isRestApiAvailable() {
+ return this == READY;
+ }
}
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconciliationUtils.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconciliationUtils.java
index 2936501a..94140c3f 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconciliationUtils.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconciliationUtils.java
@@ -379,10 +379,6 @@ public class ReconciliationUtils {
return CANCELED == status.getJobStatus().getState();
}
- public static boolean isJobCancellable(CommonStatus<?> status) {
- return RECONCILING != status.getJobStatus().getState();
- }
-
public static boolean isJobCancelling(CommonStatus<?> status) {
return status.getJobStatus() != null && CANCELLING == status.getJobStatus().getState();
}
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractJobReconciler.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractJobReconciler.java
index 3fcf8e46..9a4480d0 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractJobReconciler.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractJobReconciler.java
@@ -277,7 +277,7 @@ public abstract class AbstractJobReconciler<
if (running && savepointPossible) {
LOG.info("Using savepoint to upgrade Flink version");
return JobUpgrade.savepoint(false);
- } else if (ReconciliationUtils.isJobCancellable(resource.getStatus())) {
+ } else if (resource.getStatus().isJobCancellable()) {
LOG.info("Using last-state upgrade with cancellation to upgrade Flink version");
return JobUpgrade.lastStateUsingCancel();
} else {
@@ -354,7 +354,7 @@ public abstract class AbstractJobReconciler<
private boolean allowLastStateCancel(FlinkResourceContext<CR> ctx) {
var resource = ctx.getResource();
- if (!ReconciliationUtils.isJobCancellable(resource.getStatus())) {
+ if (!resource.getStatus().isJobCancellable()) {
return false;
}
if (resource instanceof FlinkSessionJob) {
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconciler.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconciler.java
index 180fb128..ce8baca5 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconciler.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconciler.java
@@ -88,12 +88,13 @@ public class ApplicationReconciler
var status = deployment.getStatus();
var availableUpgradeMode = super.getJobUpgrade(ctx, deployConfig);
- if (availableUpgradeMode.isAvailable() || !availableUpgradeMode.isAllowFallback()) {
+ if (availableUpgradeMode.isAvailable()) {
return availableUpgradeMode;
}
var flinkService = ctx.getFlinkService();
- if (HighAvailabilityMode.isHighAvailabilityModeActivated(deployConfig)
+ if (availableUpgradeMode.isAllowFallback()
+ && HighAvailabilityMode.isHighAvailabilityModeActivated(deployConfig)
&& HighAvailabilityMode.isHighAvailabilityModeActivated(ctx.getObserveConfig())
&& flinkService.isHaMetadataAvailable(deployConfig)) {
LOG.info(
@@ -125,7 +126,7 @@ public class ApplicationReconciler
"UpgradeFailed");
}
- return JobUpgrade.unavailable();
+ return availableUpgradeMode;
}
private void deleteJmThatNeverStarted(
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java
index f0a7b25b..18fc3690 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java
@@ -336,7 +336,7 @@ public abstract class AbstractFlinkService implements FlinkService {
savepointPath = savepointJobOrError(clusterClient, status, conf);
break;
case STATELESS:
- if (ReconciliationUtils.isJobCancellable(status)) {
+ if (status.isJobCancellable()) {
try {
cancelJobOrError(clusterClient, status, true);
} catch (Exception ex) {
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerUpgradeModeTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerUpgradeModeTest.java
index 3b50af2a..c9f70586 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerUpgradeModeTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconcilerUpgradeModeTest.java
@@ -72,6 +72,7 @@ import static org.apache.flink.api.common.JobStatus.RESTARTING;
import static org.apache.flink.api.common.JobStatus.RUNNING;
import static org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.SNAPSHOT_RESOURCE_ENABLED;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
@@ -501,6 +502,7 @@ public class ApplicationReconcilerUpgradeModeTest extends OperatorTestBase {
@ValueSource(booleans = {true, false})
public void testLastStateMaxCheckpointAge(boolean cancellable) throws Exception {
var deployment = TestUtils.buildApplicationCluster();
+ deployment.getStatus().setJobManagerDeploymentStatus(JobManagerDeploymentStatus.READY);
deployment
.getSpec()
.getFlinkConfiguration()
@@ -643,6 +645,7 @@ public class ApplicationReconcilerUpgradeModeTest extends OperatorTestBase {
jobStatus.setJobId(new JobID().toString());
// Running state, savepoint if possible
+ deployment.getStatus().setJobManagerDeploymentStatus(JobManagerDeploymentStatus.READY);
jobStatus.setState(RUNNING);
var ctx = getResourceContext(deployment);
var deployConf = ctx.getDeployConfig(deployment.getSpec());
@@ -654,6 +657,7 @@ public class ApplicationReconcilerUpgradeModeTest extends OperatorTestBase {
jobReconciler.getJobUpgrade(ctx, deployConf));
// Not running (but cancellable)
+ deployment.getStatus().setJobManagerDeploymentStatus(JobManagerDeploymentStatus.READY);
jobStatus.setState(RESTARTING);
assertEquals(
AbstractJobReconciler.JobUpgrade.lastStateUsingCancel(),
@@ -667,17 +671,26 @@ public class ApplicationReconcilerUpgradeModeTest extends OperatorTestBase {
}
private static Stream<Arguments> testLastStateCancelParams() {
- return Stream.of(
- Arguments.of(UpgradeMode.LAST_STATE, true),
- Arguments.of(UpgradeMode.LAST_STATE, false),
- Arguments.of(UpgradeMode.SAVEPOINT, true),
- Arguments.of(UpgradeMode.SAVEPOINT, false));
+ var out = new ArrayList<Arguments>();
+ for (var upgradeMode : List.of(UpgradeMode.SAVEPOINT, UpgradeMode.LAST_STATE)) {
+ for (boolean allowFallback : List.of(true, false)) {
+ for (var jmStatus : JobManagerDeploymentStatus.values()) {
+ out.add(Arguments.of(upgradeMode, allowFallback, jmStatus));
+ }
+ }
+ }
+ return out.stream();
}
@ParameterizedTest
@MethodSource("testLastStateCancelParams")
- public void testLastStateNoHaMeta(UpgradeMode upgradeMode, boolean allowFallback)
+ public void testLastStateNoHaMeta(
+ UpgradeMode upgradeMode, boolean allowFallback, JobManagerDeploymentStatus jmStatus)
throws Exception {
+ if (upgradeMode == UpgradeMode.LAST_STATE && !allowFallback) {
+ // This cannot happen
+ return;
+ }
var jobReconciler = (ApplicationReconciler) this.reconciler.getReconciler();
var deployment = TestUtils.buildApplicationCluster();
deployment
@@ -702,6 +715,8 @@ public class ApplicationReconcilerUpgradeModeTest extends OperatorTestBase {
// Set job status to running
var jobStatus = deployment.getStatus().getJobStatus();
+ deployment.getStatus().setJobManagerDeploymentStatus(jmStatus);
+
long now = System.currentTimeMillis();
jobStatus.setStartTime(Long.toString(now));
@@ -712,15 +727,21 @@ public class ApplicationReconcilerUpgradeModeTest extends OperatorTestBase {
var ctx = getResourceContext(deployment);
var deployConf = ctx.getDeployConfig(deployment.getSpec());
- if (upgradeMode == UpgradeMode.LAST_STATE) {
- assertEquals(
- AbstractJobReconciler.JobUpgrade.lastStateUsingCancel(),
- jobReconciler.getJobUpgrade(ctx, deployConf));
+ if (List.of(JobManagerDeploymentStatus.ERROR, JobManagerDeploymentStatus.MISSING)
+ .contains(jmStatus)) {
+ assertThatThrownBy(() -> jobReconciler.getJobUpgrade(ctx, deployConf))
+ .isInstanceOf(UpgradeFailureException.class)
+ .hasMessageContaining(
+ "JobManager deployment is missing and HA metadata is not available");
} else {
+ boolean immediatelyCancellable =
+ (upgradeMode == UpgradeMode.LAST_STATE || allowFallback)
+ && jmStatus == JobManagerDeploymentStatus.READY;
assertEquals(
- allowFallback
+ immediatelyCancellable
? AbstractJobReconciler.JobUpgrade.lastStateUsingCancel()
- : AbstractJobReconciler.JobUpgrade.pendingUpgrade(),
+ : new AbstractJobReconciler.JobUpgrade(
+ null, null, false, allowFallback, true),
jobReconciler.getJobUpgrade(ctx, deployConf));
}
}