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/27 15:39:01 UTC
(flink-kubernetes-operator) branch main updated: [FLINK-38867] FlinkBlueGreenDeployment used initialSavepointPath on restartSavepointNonce
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 0b14606e [FLINK-38867] FlinkBlueGreenDeployment used initialSavepointPath on restartSavepointNonce
0b14606e is described below
commit 0b14606e15fd6182d582029cf7427a7ed40e86fd
Author: James Kan <ja...@shopify.com>
AuthorDate: Thu Dec 11 12:15:22 2025 -0800
[FLINK-38867] FlinkBlueGreenDeployment used initialSavepointPath on restartSavepointNonce
Co-authored-by: Daniel Rossos <da...@shopify.com>
---
.../operator/api/bluegreen/BlueGreenDiffType.java | 8 +-
.../bluegreen/BlueGreenDeploymentService.java | 57 ++++++++-
.../diff/FlinkBlueGreenDeploymentSpecDiff.java | 21 ++-
.../operator/utils/bluegreen/BlueGreenUtils.java | 17 ++-
.../FlinkBlueGreenDeploymentControllerTest.java | 62 +++++++++
.../diff/FlinkBlueGreenDeploymentSpecDiffTest.java | 57 ++++++++-
.../utils/bluegreen/BlueGreenUtilsTest.java | 141 +++++++++++++++++++++
7 files changed, 347 insertions(+), 16 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 5581493f..e456a891 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
@@ -26,5 +26,11 @@ public enum BlueGreenDiffType {
TRANSITION,
/** Changes that only affect the child FlinkDeploymentSpec. */
- PATCH_CHILD
+ PATCH_CHILD,
+
+ /**
+ * Full redeploy from user-specified savepoint. Triggered when savepointRedeployNonce changes.
+ * Uses the initialSavepointPath from spec instead of taking a new savepoint.
+ */
+ SAVEPOINT_REDEPLOY
}
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 85de365b..132b05e2 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
@@ -42,6 +42,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.time.Instant;
+import java.util.Objects;
import static org.apache.flink.kubernetes.operator.controller.bluegreen.BlueGreenKubernetesService.deleteFlinkDeployment;
import static org.apache.flink.kubernetes.operator.controller.bluegreen.BlueGreenKubernetesService.deployCluster;
@@ -143,6 +144,25 @@ public class BlueGreenDeploymentService {
context.getDeploymentStatus().setSavepointTriggerId(null);
return markDeploymentFailing(context, error);
}
+
+ } else if (specDiff == BlueGreenDiffType.SAVEPOINT_REDEPLOY) {
+ // Savepoint redeploy: skip taking a new savepoint, use initialSavepointPath
+ var jobSpec =
+ context.getBgDeployment().getSpec().getTemplate().getSpec().getJob();
+ LOG.info(
+ "Savepoint redeploy triggered for '{}', using initialSavepointPath: {}",
+ context.getBgDeployment().getMetadata().getName(),
+ Objects.toString(jobSpec.getInitialSavepointPath(), "<none>"));
+ setLastReconciledSpec(context);
+ try {
+ return startSavepointRedeployTransition(
+ context, currentBlueGreenDeploymentType);
+ } catch (Exception e) {
+ var error =
+ "Could not start Savepoint Redeploy Transition. Details: "
+ + e.getMessage();
+ return markDeploymentFailing(context, error);
+ }
} else {
setLastReconciledSpec(context);
LOG.info(
@@ -170,6 +190,13 @@ public class BlueGreenDeploymentService {
private UpdateControl<FlinkBlueGreenDeployment> patchFlinkDeployment(
BlueGreenContext context, BlueGreenDeploymentType blueGreenDeploymentTypeToPatch) {
+ return patchFlinkDeployment(context, blueGreenDeploymentTypeToPatch, true);
+ }
+
+ private UpdateControl<FlinkBlueGreenDeployment> patchFlinkDeployment(
+ BlueGreenContext context,
+ BlueGreenDeploymentType blueGreenDeploymentTypeToPatch,
+ boolean carryOverSavepointInPatch) {
String childDeploymentName =
context.getBgDeployment().getMetadata().getName()
@@ -188,7 +215,10 @@ public class BlueGreenDeploymentService {
// will it be used by this patching? otherwise this is unnecessary, keep lastSavepoint =
// null.
Savepoint lastSavepoint =
- carryOverSavepoint(context, blueGreenDeploymentTypeToPatch, childDeploymentName);
+ carryOverSavepointInPatch
+ ? carryOverSavepoint(
+ context, blueGreenDeploymentTypeToPatch, childDeploymentName)
+ : null;
return initiateDeployment(
context,
@@ -246,6 +276,26 @@ public class BlueGreenDeploymentService {
false);
}
+ /**
+ * Starts a transition for savepoint redeploy scenario. Unlike normal transitions, this does not
+ * take a new savepoint - it uses the initialSavepointPath specified in the spec.
+ *
+ * @param context the transition context
+ * @param currentBlueGreenDeploymentType the current deployment type
+ * @return UpdateControl for the deployment
+ */
+ private UpdateControl<FlinkBlueGreenDeployment> startSavepointRedeployTransition(
+ BlueGreenContext context, BlueGreenDeploymentType currentBlueGreenDeploymentType) {
+ DeploymentTransition transition = calculateTransition(currentBlueGreenDeploymentType);
+
+ return initiateDeployment(
+ context,
+ transition.nextBlueGreenDeploymentType,
+ transition.nextState,
+ null, // Use initialSavepointPath from spec
+ false);
+ }
+
private DeploymentTransition calculateTransition(BlueGreenDeploymentType currentType) {
if (BlueGreenDeploymentType.BLUE == currentType) {
return new DeploymentTransition(
@@ -411,7 +461,10 @@ public class BlueGreenDeploymentService {
context.getDeploymentByType(oppositeDeploymentType)
.getMetadata()
.getName());
- return patchFlinkDeployment(context, oppositeDeploymentType);
+ return patchFlinkDeployment(
+ context,
+ oppositeDeploymentType,
+ diffType != BlueGreenDiffType.SAVEPOINT_REDEPLOY);
}
}
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 04c243d4..a43f3168 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
@@ -59,24 +59,31 @@ public class FlinkBlueGreenDeploymentSpecDiff {
FlinkDeploymentSpec leftSpec = left.getTemplate().getSpec();
FlinkDeploymentSpec rightSpec = right.getTemplate().getSpec();
- // Used in Case 2 & 3: Delegate to ReflectiveDiffBuilder for nested spec comparison
+ // 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 =
new ReflectiveDiffBuilder<>(deploymentMode, leftSpec, rightSpec).build();
- // Case 1: FlinkDeploymentSpecs are identical
+ // Extract the diff type from ReflectiveDiffBuilder result
+ DiffType diffType = diffResult.getType();
+
+ // Case 1: Check for savepoint redeploy first (savepointRedeployNonce changed)
+ // This takes precedence as it indicates user wants to redeploy from their
+ // initialSavepointPath
+ if (diffType == DiffType.SAVEPOINT_REDEPLOY) {
+ return BlueGreenDiffType.SAVEPOINT_REDEPLOY;
+ }
+
+ // Case 2: FlinkDeploymentSpecs are identical
if (leftSpec.equals(rightSpec)) {
return BlueGreenDiffType.IGNORE;
}
- // Extract the diff type from ReflectiveDiffBuilder result
- DiffType diffType = diffResult.getType();
-
- // Case 2: ReflectiveDiffBuilder returns IGNORE
+ // Case 3: ReflectiveDiffBuilder returns IGNORE
if (diffType == DiffType.IGNORE) {
return BlueGreenDiffType.PATCH_CHILD;
} else {
- // Case 3: ReflectiveDiffBuilder returns anything else map it to TRANSITION as well
+ // Case 4: ReflectiveDiffBuilder returns anything else map it to TRANSITION
return BlueGreenDiffType.TRANSITION;
}
}
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/bluegreen/BlueGreenUtils.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/bluegreen/BlueGreenUtils.java
index 98344a9b..27195908 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/bluegreen/BlueGreenUtils.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/bluegreen/BlueGreenUtils.java
@@ -332,7 +332,12 @@ public class BlueGreenUtils {
"spec",
FlinkBlueGreenDeploymentSpec.class);
- // The Blue/Green initialSavepointPath is only used for first-time deployments
+ // Determine which savepoint/checkpoint to restore from:
+ // 1. First-time deployments: use initialSavepointPath from spec (if set)
+ // 2. Normal transitions: use lastCheckpoint from previous deployment
+ // 3. Redeploy scenarios (lastCheckpoint is null): use initialSavepointPath from spec
+ // - savepointRedeployNonce changed
+ // - upgradeMode is STATELESS
if (isFirstDeployment) {
String initialSavepointPath =
spec.getTemplate().getSpec().getJob().getInitialSavepointPath();
@@ -346,6 +351,16 @@ public class BlueGreenUtils {
String location = lastCheckpoint.getLocation().replace("file:", "");
LOG.info("Using Blue/Green savepoint/checkpoint: " + location);
spec.getTemplate().getSpec().getJob().setInitialSavepointPath(location);
+ } else {
+ String initialSavepointPath =
+ spec.getTemplate().getSpec().getJob().getInitialSavepointPath();
+ if (initialSavepointPath != null && !initialSavepointPath.isEmpty()) {
+ LOG.info(
+ "Using user-specified initialSavepointPath for redeploy: {}",
+ initialSavepointPath);
+ } else {
+ LOG.info("Starting fresh with no savepoint restoration");
+ }
}
flinkDeployment.setSpec(spec.getTemplate().getSpec());
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 229406bb..8240b8c1 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
@@ -189,6 +189,68 @@ public class FlinkBlueGreenDeploymentControllerTest {
return rs;
}
+ @ParameterizedTest
+ @MethodSource("org.apache.flink.kubernetes.operator.TestUtils#flinkVersions")
+ public void verifySavepointRedeployNonceTriggersTransitionWithInitialSavepointPath(
+ FlinkVersion flinkVersion) throws Exception {
+ // Start with SAVEPOINT upgrade mode so normally a savepoint would be taken
+ var blueGreenDeployment =
+ buildSessionCluster(
+ TEST_DEPLOYMENT_NAME,
+ TEST_NAMESPACE,
+ flinkVersion,
+ null,
+ UpgradeMode.SAVEPOINT);
+ var rs = executeBasicDeployment(flinkVersion, blueGreenDeployment, false, null);
+
+ // Set initialSavepointPath and bump savepointRedeployNonce
+ String userSpecifiedSavepoint = "s3://bucket/my-specific-savepoint";
+ rs.deployment
+ .getSpec()
+ .getTemplate()
+ .getSpec()
+ .getJob()
+ .setInitialSavepointPath(userSpecifiedSavepoint);
+ rs.deployment.getSpec().getTemplate().getSpec().getJob().setSavepointRedeployNonce(12345L);
+ rs.deployment
+ .getSpec()
+ .getConfiguration()
+ .put(DEPLOYMENT_DELETION_DELAY.key(), String.valueOf(ALT_DELETION_DELAY_VALUE));
+ kubernetesClient.resource(rs.deployment).createOrReplace();
+
+ // Reconcile - should skip savepointing and go directly to transition
+ rs = reconcile(rs.deployment);
+
+ // Verify: Should be TRANSITIONING_TO_GREEN (not SAVEPOINTING_BLUE)
+ assertEquals(
+ FlinkBlueGreenDeploymentState.TRANSITIONING_TO_GREEN,
+ rs.reconciledStatus.getBlueGreenState());
+
+ // Verify: Green deployment should use the user-specified initialSavepointPath
+ var flinkDeployments = getFlinkDeployments();
+ assertEquals(2, flinkDeployments.size());
+ assertEquals(
+ userSpecifiedSavepoint,
+ flinkDeployments.get(1).getSpec().getJob().getInitialSavepointPath());
+
+ // Complete the transition
+ simulateSuccessfulJobStart(flinkDeployments.get(1));
+ rs = reconcile(rs.deployment);
+
+ // Wait for deletion delay
+ Thread.sleep(rs.updateControl.getScheduleDelay().get());
+ reconcile(rs.deployment);
+
+ // Verify Green is now active
+ flinkDeployments = getFlinkDeployments();
+ assertEquals(1, flinkDeployments.size());
+
+ rs = reconcile(rs.deployment);
+ assertEquals(
+ FlinkBlueGreenDeploymentState.ACTIVE_GREEN,
+ rs.reconciledStatus.getBlueGreenState());
+ }
+
@ParameterizedTest
@MethodSource("org.apache.flink.kubernetes.operator.TestUtils#flinkVersions")
public void verifyFailureBeforeTransition(FlinkVersion flinkVersion) throws Exception {
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 9fe6b8bd..272730a0 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
@@ -201,13 +201,28 @@ public class FlinkBlueGreenDeploymentSpecDiffTest {
FlinkBlueGreenDeploymentSpec spec1 = createBasicSpec();
FlinkBlueGreenDeploymentSpec spec2 = createBasicSpec();
- // Change nested spec property - setSavepointRedeployNonce triggers TRANSITION
+ // Change nested spec property - setSavepointRedeployNonce now triggers SAVEPOINT_REDEPLOY
spec2.getTemplate().getSpec().getJob().setSavepointRedeployNonce(12345L);
FlinkBlueGreenDeploymentSpecDiff diff =
new FlinkBlueGreenDeploymentSpecDiff(DEPLOYMENT_MODE, spec1, spec2);
- assertEquals(BlueGreenDiffType.TRANSITION, diff.compare());
+ assertEquals(BlueGreenDiffType.SAVEPOINT_REDEPLOY, diff.compare());
+ }
+
+ @Test
+ public void testSavepointRedeployForNonceChangeWithJarUpdate() {
+ FlinkBlueGreenDeploymentSpec spec1 = createBasicSpec();
+ FlinkBlueGreenDeploymentSpec spec2 = createBasicSpec();
+
+ // Nonce change with additional jarURI change - SAVEPOINT_REDEPLOY should still be detected
+ spec2.getTemplate().getSpec().getJob().setJarURI("local:///opt/flink/examples/other.jar");
+ spec2.getTemplate().getSpec().getJob().setSavepointRedeployNonce(12345L);
+
+ FlinkBlueGreenDeploymentSpecDiff diff =
+ new FlinkBlueGreenDeploymentSpecDiff(DEPLOYMENT_MODE, spec1, spec2);
+
+ assertEquals(BlueGreenDiffType.SAVEPOINT_REDEPLOY, diff.compare());
}
@Test
@@ -281,8 +296,8 @@ public class FlinkBlueGreenDeploymentSpecDiffTest {
FlinkBlueGreenDeploymentSpec spec2 = createBasicSpec();
// Change both top-level (configuration) and nested spec
- // With new logic, only nested spec changes matter - setSavepointRedeployNonce triggers
- // TRANSITION
+ // With new logic, only nested spec changes matter - setSavepointRedeployNonce now
+ // triggers SAVEPOINT_REDEPLOY
Map<String, String> config = new HashMap<>();
config.put("custom.config", "different-value");
spec2.setConfiguration(config);
@@ -291,7 +306,39 @@ public class FlinkBlueGreenDeploymentSpecDiffTest {
FlinkBlueGreenDeploymentSpecDiff diff =
new FlinkBlueGreenDeploymentSpecDiff(DEPLOYMENT_MODE, spec1, spec2);
- assertEquals(BlueGreenDiffType.TRANSITION, diff.compare());
+ assertEquals(BlueGreenDiffType.SAVEPOINT_REDEPLOY, diff.compare());
+ }
+
+ @Test
+ public void testSavepointRedeployTakesPrecedenceOverScale() {
+ FlinkBlueGreenDeploymentSpec spec1 = createBasicSpec();
+ FlinkBlueGreenDeploymentSpec spec2 = createBasicSpec();
+
+ // Change both nonce (SAVEPOINT_REDEPLOY) AND parallelism (SCALE)
+ // SAVEPOINT_REDEPLOY should take precedence
+ spec2.getTemplate().getSpec().getJob().setSavepointRedeployNonce(12345L);
+ spec2.getTemplate().getSpec().getJob().setParallelism(10);
+
+ FlinkBlueGreenDeploymentSpecDiff diff =
+ new FlinkBlueGreenDeploymentSpecDiff(DEPLOYMENT_MODE, spec1, spec2);
+
+ assertEquals(BlueGreenDiffType.SAVEPOINT_REDEPLOY, diff.compare());
+ }
+
+ @Test
+ public void testSavepointRedeployTakesPrecedenceOverUpgrade() {
+ FlinkBlueGreenDeploymentSpec spec1 = createBasicSpec();
+ FlinkBlueGreenDeploymentSpec spec2 = createBasicSpec();
+
+ // Change both nonce (SAVEPOINT_REDEPLOY) AND Flink version (UPGRADE)
+ // SAVEPOINT_REDEPLOY should take precedence
+ spec2.getTemplate().getSpec().getJob().setSavepointRedeployNonce(12345L);
+ spec2.getTemplate().getSpec().setFlinkVersion(FlinkVersion.v1_17);
+
+ FlinkBlueGreenDeploymentSpecDiff diff =
+ new FlinkBlueGreenDeploymentSpecDiff(DEPLOYMENT_MODE, spec1, spec2);
+
+ assertEquals(BlueGreenDiffType.SAVEPOINT_REDEPLOY, diff.compare());
}
@Test
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/bluegreen/BlueGreenUtilsTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/bluegreen/BlueGreenUtilsTest.java
index 859d19f5..a898662e 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/bluegreen/BlueGreenUtilsTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/bluegreen/BlueGreenUtilsTest.java
@@ -27,6 +27,9 @@ import org.apache.flink.kubernetes.operator.api.spec.FlinkDeploymentTemplateSpec
import org.apache.flink.kubernetes.operator.api.spec.JobSpec;
import org.apache.flink.kubernetes.operator.api.spec.UpgradeMode;
import org.apache.flink.kubernetes.operator.api.status.FlinkBlueGreenDeploymentStatus;
+import org.apache.flink.kubernetes.operator.api.status.Savepoint;
+import org.apache.flink.kubernetes.operator.api.status.SavepointFormatType;
+import org.apache.flink.kubernetes.operator.api.status.SnapshotTriggerType;
import org.apache.flink.kubernetes.operator.api.utils.SpecUtils;
import org.apache.flink.kubernetes.operator.controller.bluegreen.BlueGreenContext;
@@ -38,6 +41,9 @@ import java.util.Map;
import java.util.UUID;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
/** Tests for {@link BlueGreenUtils}. */
public class BlueGreenUtilsTest {
@@ -83,6 +89,141 @@ public class BlueGreenUtilsTest {
assertEquals(parentDeploymentName + ".jm", resultFlinkConfig.get("metrics.scope.jm"));
}
+ @Test
+ public void testSavepointRequiredBasedOnUpgradeMode() {
+ // SAVEPOINT mode requires savepoint
+ FlinkBlueGreenDeployment bgDeployment =
+ buildBlueGreenDeployment("test-app", TEST_NAMESPACE);
+ bgDeployment
+ .getSpec()
+ .getTemplate()
+ .getSpec()
+ .getJob()
+ .setUpgradeMode(UpgradeMode.SAVEPOINT);
+ BlueGreenContext context = createContext(bgDeployment);
+ assertTrue(BlueGreenUtils.isSavepointRequired(context));
+
+ // LAST_STATE mode requires savepoint
+ bgDeployment
+ .getSpec()
+ .getTemplate()
+ .getSpec()
+ .getJob()
+ .setUpgradeMode(UpgradeMode.LAST_STATE);
+ assertTrue(BlueGreenUtils.isSavepointRequired(context));
+
+ // STATELESS mode does not require savepoint
+ bgDeployment
+ .getSpec()
+ .getTemplate()
+ .getSpec()
+ .getJob()
+ .setUpgradeMode(UpgradeMode.STATELESS);
+ assertFalse(BlueGreenUtils.isSavepointRequired(context));
+ }
+
+ @Test
+ public void testPrepareFlinkDeploymentStatelessInitialSavepointPath() {
+ // Setup: STATELESS mode with initialSavepointPath set
+ FlinkBlueGreenDeployment bgDeployment =
+ buildBlueGreenDeployment("test-app", TEST_NAMESPACE);
+ bgDeployment
+ .getSpec()
+ .getTemplate()
+ .getSpec()
+ .getJob()
+ .setUpgradeMode(UpgradeMode.STATELESS);
+ bgDeployment
+ .getSpec()
+ .getTemplate()
+ .getSpec()
+ .getJob()
+ .setInitialSavepointPath("s3://bucket/savepoint-xyz");
+
+ BlueGreenContext context = createContext(bgDeployment);
+
+ // Act: Prepare deployment with null lastCheckpoint (STATELESS transition)
+ FlinkDeployment result =
+ BlueGreenUtils.prepareFlinkDeployment(
+ context,
+ BlueGreenDeploymentType.GREEN,
+ null, // No lastCheckpoint
+ false, // Not first deployment
+ bgDeployment.getMetadata());
+
+ // Assert: initialSavepointPath should be used for STATELESS
+ assertNotNull(result.getSpec().getJob().getInitialSavepointPath());
+ }
+
+ @Test
+ public void testNullLastCheckpointUsesInitialSavepointPath() {
+ // lastCheckpoint=null -> use initialSavepointPath from spec
+ FlinkBlueGreenDeployment bgDeployment =
+ buildBlueGreenDeployment("test-app", TEST_NAMESPACE);
+ bgDeployment
+ .getSpec()
+ .getTemplate()
+ .getSpec()
+ .getJob()
+ .setUpgradeMode(UpgradeMode.SAVEPOINT);
+ String initialPath = "s3://bucket/user-specified-savepoint";
+ bgDeployment
+ .getSpec()
+ .getTemplate()
+ .getSpec()
+ .getJob()
+ .setInitialSavepointPath(initialPath);
+
+ BlueGreenContext context = createContext(bgDeployment);
+
+ FlinkDeployment result =
+ BlueGreenUtils.prepareFlinkDeployment(
+ context,
+ BlueGreenDeploymentType.GREEN,
+ null, // null = nonce changed, no new savepoint taken
+ false,
+ bgDeployment.getMetadata());
+
+ assertEquals(initialPath, result.getSpec().getJob().getInitialSavepointPath());
+ }
+
+ @Test
+ public void testNormalTransitionUsesFreshSavepoint() {
+ // Normal transition → take fresh savepoint from running job → use that, not
+ // initialSavepointPath
+ FlinkBlueGreenDeployment bgDeployment =
+ buildBlueGreenDeployment("test-app", TEST_NAMESPACE);
+ bgDeployment
+ .getSpec()
+ .getTemplate()
+ .getSpec()
+ .getJob()
+ .setUpgradeMode(UpgradeMode.SAVEPOINT);
+ bgDeployment
+ .getSpec()
+ .getTemplate()
+ .getSpec()
+ .getJob()
+ .setInitialSavepointPath("s3://bucket/ignored");
+
+ BlueGreenContext context = createContext(bgDeployment);
+
+ String freshSavepoint = "s3://bucket/fresh-savepoint-from-running-job";
+ Savepoint triggered =
+ Savepoint.of(
+ freshSavepoint, SnapshotTriggerType.UPGRADE, SavepointFormatType.CANONICAL);
+
+ FlinkDeployment result =
+ BlueGreenUtils.prepareFlinkDeployment(
+ context,
+ BlueGreenDeploymentType.GREEN,
+ triggered, // Fresh savepoint provided
+ false,
+ bgDeployment.getMetadata());
+
+ assertEquals(freshSavepoint, result.getSpec().getJob().getInitialSavepointPath());
+ }
+
private static FlinkBlueGreenDeployment buildBlueGreenDeployment(
String name, String namespace) {
var deployment = new FlinkBlueGreenDeployment();