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/30 09:26:27 UTC
(flink-kubernetes-operator) branch main updated: [FLINK-38915] FlinkBlueGreenDeplomynet in place suspension handler
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 2621c9ad [FLINK-38915] FlinkBlueGreenDeplomynet in place suspension handler
2621c9ad is described below
commit 2621c9ad0884b8610caa2b15c62dbd10a26a6a53
Author: James Kan <ja...@shopify.com>
AuthorDate: Tue Jan 13 19:52:19 2026 -0800
[FLINK-38915] FlinkBlueGreenDeplomynet in place suspension handler
---
.../operator/api/bluegreen/BlueGreenDiffType.java | 14 +-
.../bluegreen/BlueGreenDeploymentService.java | 84 ++++++-
.../handlers/InitializingBlueStateHandler.java | 11 +
.../diff/FlinkBlueGreenDeploymentSpecDiff.java | 27 +++
.../FlinkBlueGreenDeploymentControllerTest.java | 242 +++++++++++++++++++++
.../diff/FlinkBlueGreenDeploymentSpecDiffTest.java | 24 ++
6 files changed, 398 insertions(+), 4 deletions(-)
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/api/bluegreen/BlueGreenDiffType.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/api/bluegreen/BlueGreenDiffType.java
index e456a891..97a83ec2 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/api/bluegreen/BlueGreenDiffType.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/api/bluegreen/BlueGreenDiffType.java
@@ -32,5 +32,17 @@ public enum BlueGreenDiffType {
* Full redeploy from user-specified savepoint. Triggered when savepointRedeployNonce changes.
* Uses the initialSavepointPath from spec instead of taking a new savepoint.
*/
- SAVEPOINT_REDEPLOY
+ SAVEPOINT_REDEPLOY,
+
+ /**
+ * In-place suspension. Triggered when job.state changes from RUNNING to SUSPENDED. Suspends the
+ * currently active child without creating a new deployment.
+ */
+ SUSPEND,
+
+ /**
+ * Resume from suspension. Triggered when job.state changes from SUSPENDED to RUNNING. Spins up
+ * the child with the current (potentially updated) spec.
+ */
+ RESUME
}
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/bluegreen/BlueGreenDeploymentService.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/bluegreen/BlueGreenDeploymentService.java
index 132b05e2..33cd868b 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/bluegreen/BlueGreenDeploymentService.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/bluegreen/BlueGreenDeploymentService.java
@@ -22,6 +22,7 @@ import org.apache.flink.kubernetes.operator.api.FlinkBlueGreenDeployment;
import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
import org.apache.flink.kubernetes.operator.api.bluegreen.BlueGreenDeploymentType;
import org.apache.flink.kubernetes.operator.api.bluegreen.BlueGreenDiffType;
+import org.apache.flink.kubernetes.operator.api.lifecycle.ResourceLifecycleState;
import org.apache.flink.kubernetes.operator.api.status.FlinkBlueGreenDeploymentState;
import org.apache.flink.kubernetes.operator.api.status.Savepoint;
import org.apache.flink.kubernetes.operator.api.status.SavepointFormatType;
@@ -110,13 +111,38 @@ public class BlueGreenDeploymentService {
*/
public UpdateControl<FlinkBlueGreenDeployment> checkAndInitiateDeployment(
BlueGreenContext context, BlueGreenDeploymentType currentBlueGreenDeploymentType) {
+
BlueGreenDiffType specDiff = getSpecDiff(context);
if (specDiff != BlueGreenDiffType.IGNORE) {
FlinkDeployment currentFlinkDeployment =
context.getDeploymentByType(currentBlueGreenDeploymentType);
- if (isFlinkDeploymentReady(currentFlinkDeployment)) {
+ if (specDiff == BlueGreenDiffType.SUSPEND && currentFlinkDeployment != null) {
+ setLastReconciledSpec(context);
+ LOG.info(
+ "In-place suspension for '{}'",
+ currentFlinkDeployment.getMetadata().getName());
+ return patchFlinkDeployment(context, currentBlueGreenDeploymentType);
+ }
+
+ if (specDiff == BlueGreenDiffType.RESUME && currentFlinkDeployment != null) {
+ setLastReconciledSpec(context);
+ LOG.info(
+ "In-place resume for '{}'", currentFlinkDeployment.getMetadata().getName());
+ return patchFlinkDeployment(context, currentBlueGreenDeploymentType);
+ }
+
+ // Check if child is currently suspended - if so, just patch specs without restart
+ if (isChildSuspended(currentFlinkDeployment)) {
+ setLastReconciledSpec(context);
+ LOG.info(
+ "Spec change while suspended for '{}'",
+ currentFlinkDeployment.getMetadata().getName());
+ return patchFlinkDeployment(context, currentBlueGreenDeploymentType);
+ }
+
+ if (currentFlinkDeployment != null && isFlinkDeploymentReady(currentFlinkDeployment)) {
if (specDiff == BlueGreenDiffType.TRANSITION) {
boolean savepointTriggered = false;
try {
@@ -173,12 +199,16 @@ public class BlueGreenDeploymentService {
} else {
if (context.getDeploymentStatus().getJobStatus().getState() != JobStatus.FAILING) {
setLastReconciledSpec(context);
+ var childName =
+ currentFlinkDeployment != null
+ ? currentFlinkDeployment.getMetadata().getName()
+ : "null";
var error =
String.format(
"Transition to %s not possible, current Flink Deployment '%s' is not READY. FAILING '%s'",
calculateTransition(currentBlueGreenDeploymentType)
.nextBlueGreenDeploymentType,
- currentFlinkDeployment.getMetadata().getName(),
+ childName,
context.getBgDeployment().getMetadata().getName());
return markDeploymentFailing(context, error);
}
@@ -188,6 +218,16 @@ public class BlueGreenDeploymentService {
return UpdateControl.noUpdate();
}
+ private boolean isChildSuspended(FlinkDeployment deployment) {
+ if (deployment == null || deployment.getSpec() == null) {
+ return false;
+ }
+ var job = deployment.getSpec().getJob();
+ return job != null
+ && job.getState()
+ == org.apache.flink.kubernetes.operator.api.spec.JobState.SUSPENDED;
+ }
+
private UpdateControl<FlinkBlueGreenDeployment> patchFlinkDeployment(
BlueGreenContext context, BlueGreenDeploymentType blueGreenDeploymentTypeToPatch) {
return patchFlinkDeployment(context, blueGreenDeploymentTypeToPatch, true);
@@ -435,6 +475,16 @@ public class BlueGreenDeploymentService {
TransitionState transitionState =
determineTransitionState(context, currentBlueGreenDeploymentType);
+ if (isChildSuspended(transitionState.nextDeployment)) {
+ if (transitionState.nextDeployment.getStatus().getLifecycleState()
+ == ResourceLifecycleState.SUSPENDED) {
+ return finalizeSuspendedDeployment(context, transitionState.nextState);
+ } else {
+ return shouldWeAbort(
+ context, transitionState.nextDeployment, transitionState.nextState);
+ }
+ }
+
if (isFlinkDeploymentReady(transitionState.nextDeployment)) {
return shouldWeDelete(
context,
@@ -447,11 +497,36 @@ public class BlueGreenDeploymentService {
}
}
+ private UpdateControl<FlinkBlueGreenDeployment> finalizeSuspendedDeployment(
+ BlueGreenContext context, FlinkBlueGreenDeploymentState nextState) {
+
+ LOG.info(
+ "Finalizing suspended deployment '{}' to {} state",
+ context.getDeploymentName(),
+ nextState);
+
+ context.getDeploymentStatus().setDeploymentReadyTimestamp(millisToInstantStr(0));
+ context.getDeploymentStatus().setAbortTimestamp(millisToInstantStr(0));
+ context.getDeploymentStatus().setSavepointTriggerId(null);
+
+ return patchStatusUpdateControl(context, nextState, JobStatus.SUSPENDED, null)
+ .rescheduleAfter(0);
+ }
+
private UpdateControl<FlinkBlueGreenDeployment> handleSpecChangesDuringTransition(
BlueGreenContext context, BlueGreenDeploymentType currentBlueGreenDeploymentType) {
if (hasSpecChanged(context)) {
BlueGreenDiffType diffType = getSpecDiff(context);
+ // Block SUSPEND during transition - wait for transition to complete first
+ if (diffType == BlueGreenDiffType.SUSPEND) {
+ LOG.info(
+ "Suspend requested during transition for '{}'. "
+ + "Waiting for transition to complete before processing suspend.",
+ context.getBgDeployment().getMetadata().getName());
+ return null;
+ }
+
if (diffType != BlueGreenDiffType.IGNORE) {
setLastReconciledSpec(context);
var oppositeDeploymentType =
@@ -658,7 +733,10 @@ public class BlueGreenDeploymentService {
context.getDeploymentStatus().setAbortTimestamp(millisToInstantStr(0));
context.getDeploymentStatus().setSavepointTriggerId(null);
- return patchStatusUpdateControl(context, nextState, JobStatus.RUNNING, null);
+ // Finalize status and reschedule immediately so any pending spec changes
+ // (e.g., suspend requested during transition) are picked up on next reconcile
+ return patchStatusUpdateControl(context, nextState, JobStatus.RUNNING, null)
+ .rescheduleAfter(0);
}
// ==================== Common Utility Methods ====================
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/bluegreen/handlers/InitializingBlueStateHandler.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/bluegreen/handlers/InitializingBlueStateHandler.java
index f2882d46..8319314f 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/bluegreen/handlers/InitializingBlueStateHandler.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/bluegreen/handlers/InitializingBlueStateHandler.java
@@ -20,6 +20,7 @@ package org.apache.flink.kubernetes.operator.controller.bluegreen.handlers;
import org.apache.flink.api.common.JobStatus;
import org.apache.flink.kubernetes.operator.api.FlinkBlueGreenDeployment;
import org.apache.flink.kubernetes.operator.api.bluegreen.BlueGreenDeploymentType;
+import org.apache.flink.kubernetes.operator.api.spec.JobState;
import org.apache.flink.kubernetes.operator.api.status.FlinkBlueGreenDeploymentState;
import org.apache.flink.kubernetes.operator.api.status.FlinkBlueGreenDeploymentStatus;
import org.apache.flink.kubernetes.operator.controller.bluegreen.BlueGreenContext;
@@ -27,6 +28,7 @@ import org.apache.flink.kubernetes.operator.controller.bluegreen.BlueGreenDeploy
import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
+import static org.apache.flink.kubernetes.operator.controller.bluegreen.BlueGreenDeploymentService.patchStatusUpdateControl;
import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenUtils.hasSpecChanged;
import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenUtils.setLastReconciledSpec;
@@ -41,6 +43,15 @@ public class InitializingBlueStateHandler extends AbstractBlueGreenStateHandler
public UpdateControl<FlinkBlueGreenDeployment> handle(BlueGreenContext context) {
FlinkBlueGreenDeploymentStatus deploymentStatus = context.getDeploymentStatus();
+ // Block initial deployment if job.state is SUSPENDED - user must start with RUNNING
+ var jobSpec = context.getBgDeployment().getSpec().getTemplate().getSpec().getJob();
+ if (jobSpec != null && jobSpec.getState() == JobState.SUSPENDED) {
+ LOG.info(
+ "Blocking initial deployment '{}' - job.state is SUSPENDED, waiting for RUNNING",
+ context.getBgDeployment().getMetadata().getName());
+ return patchStatusUpdateControl(context, null, JobStatus.SUSPENDED, null);
+ }
+
// Deploy only if this is the initial deployment (no previous spec exists)
// or if we're recovering from a failure and the spec has changed since the last attempt
if (deploymentStatus.getLastReconciledSpec() == null
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/diff/FlinkBlueGreenDeploymentSpecDiff.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/diff/FlinkBlueGreenDeploymentSpecDiff.java
index a43f3168..3e5dba77 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/diff/FlinkBlueGreenDeploymentSpecDiff.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/diff/FlinkBlueGreenDeploymentSpecDiff.java
@@ -22,6 +22,7 @@ import org.apache.flink.kubernetes.operator.api.bluegreen.BlueGreenDiffType;
import org.apache.flink.kubernetes.operator.api.diff.DiffType;
import org.apache.flink.kubernetes.operator.api.spec.FlinkBlueGreenDeploymentSpec;
import org.apache.flink.kubernetes.operator.api.spec.FlinkDeploymentSpec;
+import org.apache.flink.kubernetes.operator.api.spec.JobState;
import org.apache.flink.kubernetes.operator.api.spec.KubernetesDeploymentMode;
import lombok.NonNull;
@@ -59,6 +60,19 @@ public class FlinkBlueGreenDeploymentSpecDiff {
FlinkDeploymentSpec leftSpec = left.getTemplate().getSpec();
FlinkDeploymentSpec rightSpec = right.getTemplate().getSpec();
+ // Check for suspend/resume state changes first - these take highest precedence
+ // for in-place suspension handling
+ JobState leftJobState = getJobState(leftSpec);
+ JobState rightJobState = getJobState(rightSpec);
+
+ if (leftJobState != rightJobState) {
+ if (rightJobState == JobState.SUSPENDED) {
+ return BlueGreenDiffType.SUSPEND;
+ } else if (leftJobState == JobState.SUSPENDED && rightJobState == JobState.RUNNING) {
+ return BlueGreenDiffType.RESUME;
+ }
+ }
+
// Used in Case 2, 3 & 4: Delegate to ReflectiveDiffBuilder for nested spec comparison
// Calculate diffResult before comparison to apply in-place removal of ignored fields
DiffResult<FlinkDeploymentSpec> diffResult =
@@ -88,6 +102,19 @@ public class FlinkBlueGreenDeploymentSpecDiff {
}
}
+ /**
+ * Gets the job state from the spec, defaulting to RUNNING if not set.
+ *
+ * @param spec the FlinkDeploymentSpec
+ * @return the job state, or RUNNING if job or state is null
+ */
+ private JobState getJobState(FlinkDeploymentSpec spec) {
+ if (spec.getJob() == null || spec.getJob().getState() == null) {
+ return JobState.RUNNING;
+ }
+ return spec.getJob().getState();
+ }
+
/**
* Validates that the specs and their nested components are not null. Throws
* IllegalArgumentException if any required component is null.
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkBlueGreenDeploymentControllerTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkBlueGreenDeploymentControllerTest.java
index 11c1b931..bc571c5d 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkBlueGreenDeploymentControllerTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkBlueGreenDeploymentControllerTest.java
@@ -251,6 +251,168 @@ public class FlinkBlueGreenDeploymentControllerTest {
rs.reconciledStatus.getBlueGreenState());
}
+ @ParameterizedTest
+ @MethodSource("org.apache.flink.kubernetes.operator.TestUtils#flinkVersions")
+ public void verifySuspendAndResumeInPlace(FlinkVersion flinkVersion) throws Exception {
+ var rs = setupActiveBlueDeployment(flinkVersion);
+ assertEquals(1, getFlinkDeployments().size());
+ assertEquals(
+ FlinkBlueGreenDeploymentState.ACTIVE_BLUE,
+ rs.reconciledStatus.getBlueGreenState(),
+ "Should start as ACTIVE_BLUE");
+
+ // Suspend in-place with spec change (parallelism bump)
+ var deployment = rs.deployment;
+ deployment.getSpec().getTemplate().getSpec().getJob().setState(JobState.SUSPENDED);
+ deployment.getSpec().getTemplate().getSpec().getJob().setParallelism(5);
+ kubernetesClient.resource(deployment).createOrReplace();
+
+ rs = reconcile(deployment);
+ deployment = rs.deployment;
+ // Verify BG goes through TRANSITIONING_TO_BLUE state (unified patch mechanism)
+ assertEquals(
+ JobStatus.RECONCILING,
+ rs.reconciledStatus.getJobStatus().getState(),
+ "BG status should be RECONCILING after suspend initiated");
+ assertEquals(
+ FlinkBlueGreenDeploymentState.TRANSITIONING_TO_BLUE,
+ rs.reconciledStatus.getBlueGreenState(),
+ "Should go through TRANSITIONING_TO_BLUE during suspend");
+
+ var flinkDeployments = getFlinkDeployments();
+ assertEquals(1, flinkDeployments.size(), "Suspend should keep single child");
+ var child = flinkDeployments.get(0);
+ assertTrue(
+ child.getMetadata().getName().endsWith("-blue"), "Child should be blue deployment");
+ assertEquals(
+ JobState.SUSPENDED,
+ child.getSpec().getJob().getState(),
+ "Child should be suspended");
+ assertEquals(5, child.getSpec().getJob().getParallelism(), "Spec change should be applied");
+
+ // Simulate child becoming suspended (lifecycleState is computed from lastReconciledSpec)
+ simulateSuccessfulSuspend(child);
+ rs = reconcile(deployment);
+ deployment = rs.deployment;
+ assertEquals(JobStatus.SUSPENDED, rs.reconciledStatus.getJobStatus().getState());
+ assertEquals(
+ FlinkBlueGreenDeploymentState.ACTIVE_BLUE,
+ rs.reconciledStatus.getBlueGreenState(),
+ "Should finalize to ACTIVE_BLUE when suspended");
+
+ // Resume in-place with another spec change
+ deployment.getSpec().getTemplate().getSpec().getJob().setState(JobState.RUNNING);
+ deployment.getSpec().getTemplate().getSpec().getJob().setParallelism(6);
+ kubernetesClient.resource(deployment).createOrReplace();
+
+ rs = reconcile(deployment);
+ deployment = rs.deployment;
+ // Resume also goes through TRANSITIONING_TO_BLUE (unified patch mechanism)
+ assertEquals(
+ FlinkBlueGreenDeploymentState.TRANSITIONING_TO_BLUE,
+ rs.reconciledStatus.getBlueGreenState(),
+ "Should go through TRANSITIONING_TO_BLUE during resume");
+ // Persist the BG status update to kubernetes for the next reconcile
+ kubernetesClient.resource(deployment).updateStatus();
+
+ flinkDeployments = getFlinkDeployments();
+ assertEquals(1, flinkDeployments.size(), "Resume should keep single child");
+ child = flinkDeployments.get(0);
+ assertTrue(
+ child.getMetadata().getName().endsWith("-blue"),
+ "Child should still be blue deployment");
+ assertEquals(JobState.RUNNING, child.getSpec().getJob().getState(), "Child should resume");
+ assertEquals(6, child.getSpec().getJob().getParallelism(), "Latest spec should be applied");
+
+ // Simulate child becoming running and verify BG status syncs
+ simulateSuccessfulJobStart(child);
+ rs = reconcile(deployment);
+ assertEquals(JobStatus.RUNNING, rs.reconciledStatus.getJobStatus().getState());
+ assertEquals(
+ FlinkBlueGreenDeploymentState.ACTIVE_BLUE,
+ rs.reconciledStatus.getBlueGreenState(),
+ "Should finalize to ACTIVE_BLUE after resume");
+ }
+
+ @ParameterizedTest
+ @MethodSource("org.apache.flink.kubernetes.operator.TestUtils#flinkVersions")
+ public void verifySuspendDuringTransitionIsBlockedUntilComplete(FlinkVersion flinkVersion)
+ throws Exception {
+ // Start with a running Blue deployment with longer abort grace period
+ var blueGreenDeployment =
+ buildSessionCluster(
+ TEST_DEPLOYMENT_NAME,
+ TEST_NAMESPACE,
+ flinkVersion,
+ null,
+ UpgradeMode.STATELESS); // STATELESS to skip savepointing
+ // Set longer abort grace period BEFORE deployment to avoid abort during transition test
+ blueGreenDeployment.getSpec().getConfiguration().put(ABORT_GRACE_PERIOD.key(), "60000");
+ // Use zero deletion delay to avoid timing-based flakiness during transition completion
+ blueGreenDeployment.getSpec().getConfiguration().put(DEPLOYMENT_DELETION_DELAY.key(), "0");
+ var rs = executeBasicDeployment(flinkVersion, blueGreenDeployment, false, null);
+
+ // === TRIGGER TRANSITION ===
+ rs.deployment.getSpec().getTemplate().getSpec().getJob().setParallelism(5);
+ kubernetesClient.resource(rs.deployment).createOrReplace();
+
+ // Reconcile - start transition to GREEN
+ rs = reconcile(rs.deployment);
+
+ // GREEN deployment should be created
+ var flinkDeployments = getFlinkDeployments();
+ assertEquals(2, flinkDeployments.size());
+
+ // === SUSPEND DURING TRANSITION ===
+ rs.deployment.getSpec().getTemplate().getSpec().getJob().setState(JobState.SUSPENDED);
+ kubernetesClient.resource(rs.deployment).createOrReplace();
+
+ // Reconcile - suspend should be BLOCKED, transition continues
+ rs = reconcile(rs.deployment);
+
+ // GREEN deployment should NOT be suspended yet
+ flinkDeployments = getFlinkDeployments();
+ var greenDeployment =
+ flinkDeployments.stream()
+ .filter(fd -> fd.getMetadata().getName().endsWith("-green"))
+ .findFirst()
+ .orElseThrow();
+ assertEquals(JobState.RUNNING, greenDeployment.getSpec().getJob().getState());
+
+ // === TRANSITION COMPLETES ===
+ // Simulate GREEN becoming ready
+ simulateSuccessfulJobStart(greenDeployment);
+
+ // Reconcile to complete transition and process pending suspend
+ // (sets timestamp, deletes BLUE, finalizes to ACTIVE_GREEN, then processes suspend
+ // via TRANSITIONING_TO_GREEN)
+ for (int i = 0; i < 5; i++) {
+ rs = reconcile(rs.deployment);
+ }
+ // Persist the BG status update to kubernetes for the next reconcile
+ kubernetesClient.resource(rs.deployment).updateStatus();
+
+ // GREEN should now be suspended in spec (via patchFlinkDeployment)
+ flinkDeployments = getFlinkDeployments();
+ greenDeployment =
+ flinkDeployments.stream()
+ .filter(fd -> fd.getMetadata().getName().endsWith("-green"))
+ .findFirst()
+ .orElseThrow();
+ assertEquals(JobState.SUSPENDED, greenDeployment.getSpec().getJob().getState());
+
+ // Simulate GREEN child becoming suspended (lifecycleState is computed from
+ // lastReconciledSpec)
+ simulateSuccessfulSuspend(greenDeployment);
+
+ // Reconcile - finalizeSuspendedDeployment should set BG job status to SUSPENDED
+ rs = reconcile(rs.deployment);
+ assertEquals(JobStatus.SUSPENDED, rs.reconciledStatus.getJobStatus().getState());
+ assertEquals(
+ FlinkBlueGreenDeploymentState.ACTIVE_GREEN,
+ rs.reconciledStatus.getBlueGreenState());
+ }
+
@ParameterizedTest
@MethodSource("org.apache.flink.kubernetes.operator.TestUtils#flinkVersions")
public void verifyFailureBeforeTransition(FlinkVersion flinkVersion) throws Exception {
@@ -654,6 +816,76 @@ public class FlinkBlueGreenDeploymentControllerTest {
assertEquals(1, flinkDeployments.size());
}
+ @ParameterizedTest
+ @MethodSource("org.apache.flink.kubernetes.operator.TestUtils#flinkVersions")
+ public void verifySuspendWhenChildNotReadyPatchesWithoutFailing(FlinkVersion flinkVersion)
+ throws Exception {
+ var rs = setupActiveBlueDeployment(flinkVersion);
+
+ // Mark the active child as not ready
+ var child = getFlinkDeployments().get(0);
+ child.getStatus().getJobStatus().setState(JobStatus.RECONCILING);
+ kubernetesClient.resource(child).update();
+
+ // Suspend and bump parallelism
+ var deployment = rs.deployment;
+ deployment.getSpec().getTemplate().getSpec().getJob().setState(JobState.SUSPENDED);
+ deployment.getSpec().getTemplate().getSpec().getJob().setParallelism(9);
+ kubernetesClient.resource(deployment).createOrReplace();
+
+ rs = reconcile(deployment);
+
+ var flinkDeployments = getFlinkDeployments();
+ assertEquals(1, flinkDeployments.size(), "Should not create new child");
+ child = flinkDeployments.get(0);
+ assertEquals(JobState.SUSPENDED, child.getSpec().getJob().getState());
+ assertEquals(9, child.getSpec().getJob().getParallelism(), "Spec change should apply");
+ assertNotEquals(
+ JobStatus.FAILING,
+ rs.reconciledStatus.getJobStatus().getState(),
+ "Should not mark parent failing when suspending not-ready child");
+ }
+
+ @ParameterizedTest
+ @MethodSource("org.apache.flink.kubernetes.operator.TestUtils#flinkVersions")
+ public void verifyInitialSuspendedIsBlockedThenDeploysOnRunning(FlinkVersion flinkVersion)
+ throws Exception {
+ var bg =
+ buildSessionCluster(
+ TEST_DEPLOYMENT_NAME,
+ TEST_NAMESPACE,
+ flinkVersion,
+ null,
+ UpgradeMode.STATELESS);
+ bg.getSpec().getTemplate().getSpec().getJob().setState(JobState.SUSPENDED);
+ kubernetesClient.resource(bg).createOrReplace();
+
+ // First reconcile initializes status to INITIALIZING_BLUE
+ var rs = reconcile(bg);
+ // Second reconcile: handler detects SUSPENDED and blocks
+ rs = reconcile(rs.deployment);
+
+ // Block child creation when initial state is SUSPENDED
+ assertEquals(0, getFlinkDeployments().size(), "No child should be created");
+ assertEquals(
+ JobStatus.SUSPENDED,
+ rs.reconciledStatus.getJobStatus().getState(),
+ "Job status should be SUSPENDED when initial deployment is blocked");
+
+ // Flip to RUNNING and reconcile again
+ bg = rs.deployment;
+ bg.getSpec().getTemplate().getSpec().getJob().setState(JobState.RUNNING);
+ kubernetesClient.resource(bg).createOrReplace();
+
+ rs = reconcile(bg);
+
+ assertEquals(1, getFlinkDeployments().size(), "Child should be created after fix");
+ assertEquals(
+ JobStatus.RECONCILING,
+ rs.reconciledStatus.getJobStatus().getState(),
+ "Job status should be RECONCILING after deployment initiated");
+ }
+
// ==================== Parameterized Test Inputs ====================
static Stream<Arguments> savepointErrorProvider() {
@@ -1199,6 +1431,16 @@ public class FlinkBlueGreenDeploymentControllerTest {
kubernetesClient.resource(deployment).update();
}
+ private void simulateSuccessfulSuspend(FlinkDeployment deployment) {
+ deployment.getStatus().getJobStatus().setState(JobStatus.FINISHED);
+ deployment.getStatus().getReconciliationStatus().setState(ReconciliationState.DEPLOYED);
+ deployment
+ .getStatus()
+ .getReconciliationStatus()
+ .serializeAndSetLastReconciledSpec(deployment.getSpec(), deployment);
+ kubernetesClient.resource(deployment).update();
+ }
+
private void simulateJobFailure(FlinkDeployment deployment) {
deployment.getStatus().getJobStatus().setState(JobStatus.RECONCILING);
deployment.getStatus().getReconciliationStatus().setState(ReconciliationState.UPGRADING);
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/diff/FlinkBlueGreenDeploymentSpecDiffTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/diff/FlinkBlueGreenDeploymentSpecDiffTest.java
index 272730a0..1e60b83f 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/diff/FlinkBlueGreenDeploymentSpecDiffTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/diff/FlinkBlueGreenDeploymentSpecDiffTest.java
@@ -134,6 +134,30 @@ public class FlinkBlueGreenDeploymentSpecDiffTest {
assertEquals(BlueGreenDiffType.IGNORE, diff.compare());
}
+ @Test
+ public void testSuspendOnJobStateChange() {
+ FlinkBlueGreenDeploymentSpec spec1 = createBasicSpec(); // RUNNING default
+ FlinkBlueGreenDeploymentSpec spec2 = createBasicSpec();
+ spec2.getTemplate().getSpec().getJob().setState(JobState.SUSPENDED);
+
+ FlinkBlueGreenDeploymentSpecDiff diff =
+ new FlinkBlueGreenDeploymentSpecDiff(DEPLOYMENT_MODE, spec1, spec2);
+
+ assertEquals(BlueGreenDiffType.SUSPEND, diff.compare());
+ }
+
+ @Test
+ public void testResumeOnJobStateChange() {
+ FlinkBlueGreenDeploymentSpec spec1 = createBasicSpec();
+ spec1.getTemplate().getSpec().getJob().setState(JobState.SUSPENDED);
+ FlinkBlueGreenDeploymentSpec spec2 = createBasicSpec(); // RUNNING default
+
+ FlinkBlueGreenDeploymentSpecDiff diff =
+ new FlinkBlueGreenDeploymentSpecDiff(DEPLOYMENT_MODE, spec1, spec2);
+
+ assertEquals(BlueGreenDiffType.RESUME, diff.compare());
+ }
+
@Test
public void testIgnoreForRootPodTemplateAdditionalProps() {
FlinkBlueGreenDeploymentSpec spec1 = createBasicSpec();