Fix FBGD stale-state bugs on aborted transitions (#3)

jhung-lyft · jennifer-xiong25 · web-flow · commit 1d4b5e536e78 · 2026-05-22T13:58:43.000-04:00
## TL;DR Fixes three bugs in `FlinkBlueGreenDeployment` that compound to leave a transition silently restoring from a stale savepoint: | # | Bug | Fix | |---|-----|-----| | 1 | `savepointTriggerId` not cleared on abort → next deploy stuck FAILING+IGNORE | Clear it in `abortDeployment` (mirrors upstream [a00bf27](apache@a00bf27)) | | 2 | `setLastReconciledSpec` runs before `startTransition` → failed spec marked reconciled, retries no-op | Stamp only on success (mirrors upstream [apache#1072](apache#1072)) | | 3 | Suspended child carries stale `status.upgradeSavepointPath` → JM restores from old savepoint on retry | Delete the failed child in `abortDeployment` so the next transition fresh-creates it | Verified in unit test and end-to-end on `data-stg` — see [Validation](#Staging-validation-on-data-stg) below. ## Issue 1 — `savepointTriggerId` not cleared on abort `abortDeployment` left `status.savepointTriggerId` in place. The next reconcile tried to re-fetch the trigger from the active side's JM, which had TTL-evicted it (~300s), threw `Could not fetch savepoint with triggerId: X`, and left FBGD stuck in FAILING + IGNORE. Required a `restartNonce` bump to escape. ## Issue 2 — `setLastReconciledSpec` stamped before `startTransition` returns In `checkAndInitiateDeployment`, `setLastReconciledSpec` ran *before* `startTransition`. If `startTransition` threw (e.g., from Issue 1), the failed spec was already marked reconciled, so the next reconcile saw `specDiff = IGNORE` and silently did nothing until another spec change came in. ## Issue 3 — Stale `status.upgradeSavepointPath` shadows fresh `spec.initialSavepointPath` `AbstractJobReconciler.restoreJob()` reads `status.jobStatus.upgradeSavepointPath`, not `spec.initialSavepointPath`, in the `terminal && !isHaMetadataAvailable` upgrade branch. A suspended child retained its stale `upgradeSavepointPath` from the previous failed deploy. On retry, FBGD wrote a fresh `spec.initialSavepointPath`, but `createOrReplace` does not touch the status subresource — so the stale status value won and the JM restored from an arbitrarily old savepoint while the transition appeared successful. Upstream [apache#1073](apache#1073) does NOT fix this: its `!isHaMetadataAvailable || isJmAccessible` gate is satisfied with empty HA, so the operator still routes through the stale-status path. **Fix**: `abortDeployment` now deletes the failed child instead of suspending it. The next transition fresh-creates the child via `createOrReplace`'s POST path, with no status to be stale. Falls back to `suspendFlinkDeployment` if the delete call fails, to avoid an inconsistent state. ## Tests - `verifyFailedChildDeletedOnAbort` (new): forces a failed transition, asserts the failed child is gone, then asserts the retry fresh-creates GREEN with the fresh savepoint in spec and no carry-over status. - `verifyFailureDuringTransition` / `verifyFailureBeforeFirstDeployment`: updated to assert the failed child is absent (not SUSPENDED) after abort. - `verifySavepointFetchFailureRecovery`: asserts the failed spec is NOT present in `lastReconciledSpec` (Issue 2). ## Test plan - [x] `FlinkBlueGreenDeploymentControllerTest` → 60/60 pass - [x] `*ControllerTest` → 148/148 pass (no regressions in FlinkDeployment / SessionJob / StateSnapshot) - [x] Deployed to `data-stg`; forced abort + retry on `beamperfk8s1`; verified all three fixes ([details below](#Staging-validation-on-data-stg)) ## Staging validation on data-stg Forced a failed transition on `beamperfk8s1` with a bad image tag, then retried with a good image. | Time (UTC) | Event | |------------|-------| | 15:08:40 | First transition: triggerId `5d75a09efb99...`, savepoint `6966a02076a5` → new GREEN | | 15:09:42 | `Aborting deployment 'beamperfk8s1-green', rolling B/G deployment back to ACTIVE_BLUE` | | 15:09:44 | GREEN audit: `Status[Job] \| Error \| DELETED \| ErrImagePull` | | 15:11:40 | Retry: NEW triggerId `5f1d597953fe...` (different from first) | | 15:11:55 | Fresh savepoint `b0d173acab2e` → fresh GREEN | | 15:12:56 | GREEN JM log: `Starting job ... from savepoint .../savepoint-4d93e8-b0d173acab2e` | | 15:13:23 | Old BLUE deleted (normal post-transition cleanup) | - **1**: retry used a new triggerId, not the TTL-evicted first one - **2**: retry transition actually ran (would have been IGNORE'd if the failed spec had been stamped) - **3**: failed GREEN was DELETED (not suspended); fresh GREEN's JM restored from the fresh `b0d173acab2e`, not the prior attempt's `6966a02076a5` ## Production incident this resolves May 7 `demandingestion`: a transient deploy failure left stale `savepointTriggerId` and `status.upgradeSavepointPath` on a suspended GREEN; the failed spec was stamped as reconciled, freezing retries; when the trigger TTL evicted and a deploy went through, the JM silently restored from a 9-day-old savepoint. Full RCA: [Deploy Failure Investigation](https://docs.google.com/document/d/1rwN2TABe-3VtkHWBttLbLlW_P6Q-TMs0cuAs6fd-cQA/edit). --------- Co-authored-by: jennifer-xiong25 <jennifer.xiong@shopify.com>
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
@@ -134,10 +134,18 @@ public UpdateControl<FlinkBlueGreenDeployment> checkAndInitiateDeployment(
                                 .rescheduleAfter(getReconciliationReschedInterval(context));
                     }
 
-                    setLastReconciledSpec(context);
                     try {
-                        return startTransition(
-                                context, currentBlueGreenDeploymentType, currentFlinkDeployment);
+                        var result =
+                                startTransition(
+                                        context,
+                                        currentBlueGreenDeploymentType,
+                                        currentFlinkDeployment);
+                        // Only stamp lastReconciledSpec after the transition
+                        // succeeds. If stamped before and the transition
+                        // fails/aborts, lastReconciledSpec drifts from the
+                        // active child's actual spec
+                        setLastReconciledSpec(context);
+                        return result;
                     } catch (Exception e) {
                         var error = "Could not start Transition. Details: " + e.getMessage();
                         context.getDeploymentStatus().setSavepointTriggerId(null);
@@ -551,11 +559,24 @@ private UpdateControl<FlinkBlueGreenDeployment> abortDeployment(
             FlinkBlueGreenDeploymentState nextState,
             String deploymentName) {
 
-        suspendFlinkDeployment(context, nextDeployment);
+        // Delete the failed child rather than suspending it. Keeping a suspended FD around leaves
+        // its status subresource (notably status.jobStatus.upgradeSavepointPath) on the cluster.
+        // createOrReplace on the next transition writes a fresh spec but does not touch status,
+        // so AbstractJobReconciler.restoreJob() would read the stale savepoint path and restore
+        // the job from it. Deleting forces the next transition to use the first-deployment code
+        // path, which reads spec.initialSavepointPath correctly.
+        boolean deleted = deleteFlinkDeployment(nextDeployment, context);
+        if (!deleted) {
+            LOG.warn(
+                    "Failed to delete child '{}' during abort; falling back to suspend",
+                    deploymentName);
+            suspendFlinkDeployment(context, nextDeployment);
+        }
 
         FlinkBlueGreenDeploymentState previousState =
                 getPreviousState(nextState, context.getDeployments());
         context.getDeploymentStatus().setBlueGreenState(previousState);
+        context.getDeploymentStatus().setSavepointTriggerId(null);
 
         var error =
                 String.format(
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
@@ -297,25 +297,24 @@ public void verifyFailureDuringTransition(FlinkVersion flinkVersion) throws Exce
 
         assertTrue(rs.updateControl.isPatchStatus());
 
-        // The first job should be RUNNING, the second should be SUSPENDED
         assertFailingJobStatus(rs);
         // No longer TRANSITIONING_TO_GREEN and rolled back to ACTIVE_BLUE
         assertEquals(
                 FlinkBlueGreenDeploymentState.ACTIVE_BLUE, rs.reconciledStatus.getBlueGreenState());
+        // The failed child is deleted on abort (instead of suspended), so only the active
+        // BLUE deployment should remain. Deletion prevents stale status.upgradeSavepointPath
+        // on the failed child from being read by the next transition's restoreJob().
         var flinkDeployments = getFlinkDeployments();
-        assertEquals(2, flinkDeployments.size());
+        assertEquals(1, flinkDeployments.size());
         assertEquals(
                 JobStatus.RUNNING, flinkDeployments.get(0).getStatus().getJobStatus().getState());
         assertEquals(
                 ReconciliationState.DEPLOYED,
                 flinkDeployments.get(0).getStatus().getReconciliationStatus().getState());
-        // The B/G controller changes the State = SUSPENDED, the actual suspension is done by the
-        // FlinkDeploymentController
-        assertEquals(JobState.SUSPENDED, flinkDeployments.get(1).getSpec().getJob().getState());
-        assertEquals(
-                ReconciliationState.UPGRADING,
-                flinkDeployments.get(1).getStatus().getReconciliationStatus().getState());
         assertTrue(instantStrToMillis(rs.reconciledStatus.getAbortTimestamp()) > 0);
+        // savepointTriggerId must be cleared on abort so the next transition
+        // triggers a fresh savepoint instead of reusing a stale triggerId
+        assertNull(rs.reconciledStatus.getSavepointTriggerId());
 
         // Simulate another change in the spec to trigger a redeployment
         customValue = UUID.randomUUID().toString();
@@ -325,6 +324,96 @@ public void verifyFailureDuringTransition(FlinkVersion flinkVersion) throws Exce
         testTransitionToGreen(rs, customValue, null);
     }
 
+    /**
+     * abortDeployment deletes the failed child FD instead of suspending it. Keeping a suspended
+     * child around leaves its status subresource (notably status.jobStatus.upgradeSavepointPath)
+     * intact, which AbstractJobReconciler.restoreJob() then reads on the next transition and uses
+     * to restore the job — silently shadowing the fresh spec.initialSavepointPath that FBGD writes.
+     *
+     * <p>This test asserts that after an aborted transition the failed child is gone, and the next
+     * transition fresh-creates it with the new savepoint.
+     */
+    @ParameterizedTest
+    @MethodSource("org.apache.flink.kubernetes.operator.TestUtils#flinkVersions")
+    public void verifyFailedChildDeletedOnAbort(FlinkVersion flinkVersion) throws Exception {
+        var blueGreenDeployment =
+                buildSessionCluster(
+                        TEST_DEPLOYMENT_NAME,
+                        TEST_NAMESPACE,
+                        flinkVersion,
+                        null,
+                        UpgradeMode.SAVEPOINT);
+
+        var abortGracePeriodMs = 1200;
+        var reschedulingIntervalMs = 3000;
+        blueGreenDeployment
+                .getSpec()
+                .getConfiguration()
+                .put(ABORT_GRACE_PERIOD.key(), String.valueOf(abortGracePeriodMs));
+        blueGreenDeployment
+                .getSpec()
+                .getConfiguration()
+                .put(
+                        RECONCILIATION_RESCHEDULING_INTERVAL.key(),
+                        String.valueOf(reschedulingIntervalMs));
+
+        // 1. Initial deploy → ACTIVE_BLUE
+        var rs = executeBasicDeployment(flinkVersion, blueGreenDeployment, false, null);
+        String blueName = getFlinkDeployments().get(0).getMetadata().getName();
+
+        // 2. Spec change → savepoint → start transition to GREEN
+        simulateChangeInSpec(rs.deployment, UUID.randomUUID().toString(), 0, null);
+        rs = handleSavepoint(rs);
+        rs = reconcile(rs.deployment);
+        assertEquals(
+                FlinkBlueGreenDeploymentState.TRANSITIONING_TO_GREEN,
+                rs.reconciledStatus.getBlueGreenState());
+        assertEquals(2, getFlinkDeployments().size());
+
+        // 3. GREEN never becomes ready → abort
+        Long reschedDelayMs = 0L;
+        for (int i = 0; i < 2; i++) {
+            rs = reconcile(rs.deployment);
+            reschedDelayMs = rs.updateControl.getScheduleDelay().get();
+        }
+        Thread.sleep(reschedDelayMs);
+        rs = reconcile(rs.deployment);
+        assertFailingJobStatus(rs);
+        assertEquals(
+                FlinkBlueGreenDeploymentState.ACTIVE_BLUE, rs.reconciledStatus.getBlueGreenState());
+
+        // 4. The failed child must be deleted (not suspended). Only BLUE remains.
+        var remaining = getFlinkDeployments();
+        assertEquals(1, remaining.size(), "failed child should be deleted on abort");
+        assertEquals(blueName, remaining.get(0).getMetadata().getName());
+
+        // 5. Second transition: spec change → fresh savepoint → GREEN is fresh-created.
+        simulateChangeInSpec(rs.deployment, UUID.randomUUID().toString(), 0, null);
+        rs = handleSavepoint(rs);
+        rs = reconcile(rs.deployment);
+        assertEquals(
+                FlinkBlueGreenDeploymentState.TRANSITIONING_TO_GREEN,
+                rs.reconciledStatus.getBlueGreenState());
+
+        var greenAfterRetry =
+                getFlinkDeployments().stream()
+                        .filter(d -> !d.getMetadata().getName().equals(blueName))
+                        .findFirst()
+                        .orElseThrow(
+                                () -> new AssertionError("GREEN should be fresh-created on retry"));
+        // Freshly-created FD must start with no status, so restoreJob() falls into the first-
+        // deployment branch and uses spec.initialSavepointPath rather than status.
+        assertTrue(
+                greenAfterRetry.getStatus() == null
+                        || greenAfterRetry.getStatus().getJobStatus() == null
+                        || greenAfterRetry.getStatus().getJobStatus().getUpgradeSavepointPath()
+                                == null,
+                "freshly created GREEN must not carry status.upgradeSavepointPath");
+        assertNotNull(
+                greenAfterRetry.getSpec().getJob().getInitialSavepointPath(),
+                "GREEN must have a fresh initialSavepointPath in spec");
+    }
+
     private static String getFlinkConfigurationValue(
             FlinkDeploymentSpec flinkDeploymentSpec, String propertyName) {
         return flinkDeploymentSpec.getFlinkConfiguration().get(propertyName).asText();
@@ -389,11 +478,9 @@ public void verifyFailureBeforeFirstDeployment(FlinkVersion flinkVersion) throws
         assertEquals(
                 FlinkBlueGreenDeploymentState.INITIALIZING_BLUE,
                 rs.reconciledStatus.getBlueGreenState());
+        // The failed child is deleted on abort.
         var flinkDeployments = getFlinkDeployments();
-        assertEquals(1, flinkDeployments.size());
-        // The B/G controller changes the State = SUSPENDED, the actual suspension is done by the
-        // FlinkDeploymentController
-        assertEquals(JobState.SUSPENDED, flinkDeployments.get(0).getSpec().getJob().getState());
+        assertEquals(0, flinkDeployments.size());
 
         // No-op if the spec remains the same
         rs = reconcile(rs.deployment);
@@ -537,6 +624,13 @@ public void verifySavepointFetchFailureRecovery(FlinkVersion flinkVersion) throw
         rs = reconcile(rs.deployment);
         assertFailingWithError(rs, "Could not start Transition", error);
 
+        // lastReconciledSpec must NOT contain the failed spec change — otherwise
+        // subsequent deploys see no diff and fall into PATCH_CHILD (in-place
+        // upgrade) instead of a proper B/G transition
+        assertFalse(
+                rs.reconciledStatus.getLastReconciledSpec().contains(customValue),
+                "lastReconciledSpec should not be stamped when transition fails");
+
         // Recovery: Clear the fetch error and try again with new spec change
         flinkService.clearSavepointFetchError();
         customValue = UUID.randomUUID().toString() + "_recovery";
