Skip to content

Revert FlinkBlueGreenDeployment lastReconciledSpec back to old spec during abort workflow #26

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jennifer-xiong25 merged 3 commits into from
Mar 27, 2026
Merged

Revert FlinkBlueGreenDeployment lastReconciledSpec back to old spec during abort workflow #26

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
7a353a1
new fixes
jennifer-xiong25 Mar 20, 2026
2bf72e2
new fixes
jennifer-xiong25 Mar 20, 2026
7c594f9
new fixes
jennifer-xiong25 Mar 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 32 additions & 4 deletions ...org/apache/flink/kubernetes/operator/controller/bluegreen/BlueGreenDeploymentService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,9 @@
import org.slf4j.LoggerFactory;

import java.time.Instant;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;

import static org.apache.flink.kubernetes.operator.controller.bluegreen.BlueGreenKubernetesService.deleteFlinkDeployment;
import static org.apache.flink.kubernetes.operator.controller.bluegreen.BlueGreenKubernetesService.deployCluster;
Expand All @@ -58,6 +60,7 @@
import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenUtils.isSavepointRequired;
import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenUtils.millisToInstantStr;
import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenUtils.prepareFlinkDeployment;
import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenUtils.revertToLastSpec;
import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenUtils.setLastReconciledSpec;
import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenUtils.triggerSavepoint;

Expand All @@ -67,6 +70,15 @@ public class BlueGreenDeploymentService {
private static final Logger LOG = LoggerFactory.getLogger(BlueGreenDeploymentService.class);
private static final long RETRY_DELAY_MS = 500;

/**
* In-memory cache of the pre-transition lastReconciledSpec, keyed by namespace. Saved before
* setLastReconciledSpec overwrites it during a transition, restored on abort so
* lastReconciledSpec stays consistent with the active child. Lost on operator restart, which
* falls back to pre-fix behavior (no revert) — acceptable since restart during the abort window
* is rare and not a regression.
*/
private final Map<String, String> previousReconciledSpecs = new ConcurrentHashMap<>();

// ==================== Deployment Initiation Methods ====================

/**
Expand Down Expand Up @@ -163,15 +175,13 @@ public UpdateControl<FlinkBlueGreenDeployment> checkAndInitiateDeployment(
}

try {
// Save pre-transition lastReconciledSpec so we can restore on abort
savePreTransitionSpec(context);
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) {
Expand All @@ -189,6 +199,8 @@ public UpdateControl<FlinkBlueGreenDeployment> checkAndInitiateDeployment(
context.getBgDeployment().getMetadata().getName(),
Objects.toString(jobSpec.getInitialSavepointPath(), "<none>"));
try {
// Save pre-transition lastReconciledSpec so we can restore on abort
savePreTransitionSpec(context);
var result =
startSavepointRedeployTransition(
context, currentBlueGreenDeploymentType);
Expand Down Expand Up @@ -697,13 +709,28 @@ private UpdateControl<FlinkBlueGreenDeployment> abortDeployment(
context.getDeploymentStatus().setBlueGreenState(previousState);
context.getDeploymentStatus().setSavepointTriggerId(null);

// Restore lastReconciledSpec and the B/G CR spec to the pre-transition state
// so they stay consistent with the active child that is still running.
String namespace = context.getBgDeployment().getMetadata().getNamespace();

@james-kan-shopify james-kan-shopify Mar 26, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

call out: if the BG gets deleted or we exit somewhere during the transition after failure, or without failure we don't seem to remove the entry in the previousReconciledSpecs at that point. Is that a risk? it's only of size 1 per namespace but do we want to keep it around, is there any chance we could inadvertently end up using the value in there resulting in an improper restore somehow? (i.e. we end up using the stale one to restore)

@jennifer-xiong25 jennifer-xiong25 Mar 26, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

it's not a risk because savePreTransitionSpec always overwrites the entry for that namespace before each transition. By the time abortDeployment runs, the map has the correct pre-transition spec for the current transition, not a stale one from a previous transition.

@james-kan-shopify james-kan-shopify Mar 27, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

right beginning of each round will wipe it out. okay perfect, thanks for talking through these with me!

String previousSpec = previousReconciledSpecs.remove(namespace);
if (previousSpec != null) {
context.getDeploymentStatus().setLastReconciledSpec(previousSpec);
revertToLastSpec(context);
}

var error =
String.format(
"Aborting deployment '%s', rolling B/G deployment back to %s",
deploymentName, previousState);
return markDeploymentFailing(context, error);
}

private void savePreTransitionSpec(BlueGreenContext context) {
String namespace = context.getBgDeployment().getMetadata().getNamespace();
previousReconciledSpecs.put(
namespace, context.getDeploymentStatus().getLastReconciledSpec());
}

@NotNull
private static UpdateControl<FlinkBlueGreenDeployment> markDeploymentFailing(
BlueGreenContext context, String error) {
Expand Down Expand Up @@ -744,6 +771,7 @@ public UpdateControl<FlinkBlueGreenDeployment> finalizeBlueGreenDeployment(
context.getDeploymentStatus().setDeploymentReadyTimestamp(millisToInstantStr(0));
context.getDeploymentStatus().setAbortTimestamp(millisToInstantStr(0));
context.getDeploymentStatus().setSavepointTriggerId(null);
previousReconciledSpecs.remove(context.getBgDeployment().getMetadata().getNamespace());

updateBlueGreenIngress(context, nextState);

Expand Down
6 changes: 6 additions & 0 deletions ...g/apache/flink/kubernetes/operator/controller/FlinkBlueGreenDeploymentControllerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -549,6 +549,12 @@ public void verifyFailureDuringTransition(FlinkVersion flinkVersion) throws Exce
// savepointTriggerId must be cleared on abort so the next transition
// triggers a fresh savepoint instead of reusing a stale triggerId
assertNull(rs.reconciledStatus.getSavepointTriggerId());
// lastReconciledSpec must NOT contain the failed spec change — on abort
// it should be reverted to the pre-transition spec so it stays consistent
// with the active child that is still running
assertFalse(
rs.reconciledStatus.getLastReconciledSpec().contains(customValue),
"lastReconciledSpec should be reverted on abort to match the active child");

// Simulate another change in the spec to trigger a redeployment
customValue = UUID.randomUUID().toString();
Expand Down
Loading