From e7eda708a5246cd2d5c9255f19640dc67a4a7065 Mon Sep 17 00:00:00 2001 From: James Kan Date: Tue, 10 Feb 2026 16:01:16 -0800 Subject: [PATCH 1/5] Flinkbluegreendeployment autoscaling --- .../bluegreen/BlueGreenDeploymentService.java | 290 ++++++++++-- .../bluegreen/BlueGreenKubernetesService.java | 25 ++ .../AbstractFlinkResourceReconciler.java | 54 ++- .../utils/bluegreen/BlueGreenUtils.java | 14 + ...linkBlueGreenDeploymentControllerTest.java | 425 ++++++++++++++++++ 5 files changed, 775 insertions(+), 33 deletions(-) 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 9543fecabd..a5e296b56d 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 @@ -23,10 +23,13 @@ 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.spec.FlinkBlueGreenDeploymentSpec; +import org.apache.flink.kubernetes.operator.api.spec.FlinkDeploymentSpec; 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; import org.apache.flink.kubernetes.operator.api.status.SnapshotTriggerType; +import org.apache.flink.kubernetes.operator.api.utils.SpecUtils; import org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions; import org.apache.flink.kubernetes.operator.controller.FlinkBlueGreenDeployments; import org.apache.flink.kubernetes.operator.controller.FlinkResourceContext; @@ -44,6 +47,8 @@ import org.slf4j.LoggerFactory; import java.time.Instant; +import java.util.HashMap; +import java.util.Map; import java.util.Objects; import static org.apache.flink.kubernetes.operator.controller.bluegreen.BlueGreenKubernetesService.deleteFlinkDeployment; @@ -53,7 +58,6 @@ import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenUtils.fetchSavepointInfo; import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenUtils.getReconciliationReschedInterval; import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenUtils.getSpecDiff; -import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenUtils.hasSpecChanged; import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenUtils.instantStrToMillis; import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenUtils.isSavepointRequired; import static org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenUtils.millisToInstantStr; @@ -113,6 +117,7 @@ public UpdateControl initiateDeployment( public UpdateControl checkAndInitiateDeployment( BlueGreenContext context, BlueGreenDeploymentType currentBlueGreenDeploymentType) { + // Check user spec changes first (takes precedence over autoscaler changes) BlueGreenDiffType specDiff = getSpecDiff(context); if (specDiff != BlueGreenDiffType.IGNORE) { @@ -150,7 +155,7 @@ public UpdateControl checkAndInitiateDeployment( savepointTriggered = handleSavepoint(context, currentFlinkDeployment); } catch (Exception e) { var error = "Could not trigger Savepoint. Details: " + e.getMessage(); - return markDeploymentFailing(context, error); + return markDeploymentFailing(context, error, e); } if (savepointTriggered) { @@ -169,7 +174,7 @@ public UpdateControl checkAndInitiateDeployment( } catch (Exception e) { var error = "Could not start Transition. Details: " + e.getMessage(); context.getDeploymentStatus().setSavepointTriggerId(null); - return markDeploymentFailing(context, error); + return markDeploymentFailing(context, error, e); } } else if (specDiff == BlueGreenDiffType.SAVEPOINT_REDEPLOY) { @@ -188,7 +193,7 @@ public UpdateControl checkAndInitiateDeployment( var error = "Could not start Savepoint Redeploy Transition. Details: " + e.getMessage(); - return markDeploymentFailing(context, error); + return markDeploymentFailing(context, error, e); } } else { setLastReconciledSpec(context); @@ -216,6 +221,20 @@ public UpdateControl checkAndInitiateDeployment( } } + // No user spec changes — check if autoscaler produced overrides on the active child. + // If so, trigger a B/G transition with those overrides applied to the new child. + // Skip when FAILING to prevent an infinite loop: abort sets FAILING and rolls back + // to ACTIVE, but the old child still has the same autoscaler overrides → re-detected + // → new transition → abort → re-detected → ... A user spec change clears FAILING + // via the normal transition path above (sets RECONCILING in initiateDeployment). + if (context.getDeploymentStatus().getJobStatus().getState() != JobStatus.FAILING) { + var autoscalerResult = + checkAutoscalerTriggeredTransition(context, currentBlueGreenDeploymentType); + if (autoscalerResult != null) { + return autoscalerResult; + } + } + return UpdateControl.noUpdate(); } @@ -307,6 +326,13 @@ private UpdateControl startTransition( FlinkDeployment currentFlinkDeployment) { DeploymentTransition transition = calculateTransition(currentBlueGreenDeploymentType); + LOG.info( + "Starting Blue/Green transition for '{}': {} -> {} (from child '{}')", + context.getDeploymentName(), + currentBlueGreenDeploymentType, + transition.nextBlueGreenDeploymentType, + currentFlinkDeployment.getMetadata().getName()); + Savepoint lastCheckpoint = configureInitialSavepoint(context, currentFlinkDeployment); return initiateDeployment( @@ -436,7 +462,7 @@ private boolean handleSavepoint( if (savepointTriggerId == null || savepointTriggerId.isEmpty()) { String triggerId = triggerSavepoint(ctx); - LOG.info("Savepoint requested (triggerId: {}", triggerId); + LOG.info("Savepoint requested (triggerId: {})", triggerId); context.getDeploymentStatus().setSavepointTriggerId(triggerId); return true; } @@ -516,35 +542,32 @@ private UpdateControl finalizeSuspendedDeployment( private UpdateControl 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; - } + // Parent spec is never persisted with autoscaler values, + // so getSpecDiff detects only user-initiated changes. + BlueGreenDiffType diffType = getSpecDiff(context); - if (diffType != BlueGreenDiffType.IGNORE) { - setLastReconciledSpec(context); - var oppositeDeploymentType = - context.getOppositeDeploymentType(currentBlueGreenDeploymentType); - LOG.info( - "Patching FlinkDeployment '{}' during handleSpecChangesDuringTransition", - context.getDeploymentByType(oppositeDeploymentType) - .getMetadata() - .getName()); - return patchFlinkDeployment( - context, - oppositeDeploymentType, - diffType != BlueGreenDiffType.SAVEPOINT_REDEPLOY); - } + if (diffType == BlueGreenDiffType.IGNORE) { + return null; } - return null; + // 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; + } + + setLastReconciledSpec(context); + var oppositeDeploymentType = + context.getOppositeDeploymentType(currentBlueGreenDeploymentType); + LOG.info( + "Patching FlinkDeployment '{}' during handleSpecChangesDuringTransition", + context.getDeploymentByType(oppositeDeploymentType).getMetadata().getName()); + return patchFlinkDeployment( + context, oppositeDeploymentType, diffType != BlueGreenDiffType.SAVEPOINT_REDEPLOY); } private TransitionState determineTransitionState( @@ -632,9 +655,11 @@ private UpdateControl deleteDeployment( boolean deleted = deleteFlinkDeployment(currentDeployment, context); if (!deleted) { - LOG.info("FlinkDeployment '{}' not deleted, will retry", currentDeployment); + LOG.warn( + "FlinkDeployment '{}' not deleted, will retry", + currentDeployment.getMetadata().getName()); } else { - LOG.info("FlinkDeployment '{}' deleted!", currentDeployment); + LOG.info("FlinkDeployment '{}' deleted!", currentDeployment.getMetadata().getName()); } return UpdateControl.noUpdate().rescheduleAfter(RETRY_DELAY_MS); @@ -682,6 +707,12 @@ private UpdateControl abortDeployment( suspendFlinkDeployment(context, nextDeployment); + // Clear stale savepointTriggerId to prevent reuse after abort (defense-in-depth). + context.getDeploymentStatus().setSavepointTriggerId(null); + + // Autoscaler overrides persist in the active child's lastReconciledSpec + // and will be re-detected on the next ACTIVE reconcile. + FlinkBlueGreenDeploymentState previousState = getPreviousState(nextState, context.getDeployments()); context.getDeploymentStatus().setBlueGreenState(previousState); @@ -700,6 +731,13 @@ private static UpdateControl markDeploymentFailing( return patchStatusUpdateControl(context, null, JobStatus.FAILING, error); } + @NotNull + private static UpdateControl markDeploymentFailing( + BlueGreenContext context, String error, Throwable cause) { + LOG.error(error, cause); + return patchStatusUpdateControl(context, null, JobStatus.FAILING, error); + } + private static FlinkBlueGreenDeploymentState getPreviousState( FlinkBlueGreenDeploymentState nextState, FlinkBlueGreenDeployments deployments) { FlinkBlueGreenDeploymentState previousState; @@ -812,6 +850,194 @@ public void updateBlueGreenIngress( blueGreenContext.getJosdkContext()); } + // ==================== Autoscaler Detection for Blue/Green ==================== + + /** + * Detects autoscaler overrides on the active child and initiates a B/G transition if any are + * found. Overrides are applied ephemerally to the parent's in-memory spec so the new child + * inherits them via {@code prepareFlinkDeployment}; the parent spec is restored in a finally + * block to prevent phantom diffs. + * + * @return UpdateControl if a transition was initiated, or {@code null} if no overrides + */ + @Nullable + private UpdateControl checkAutoscalerTriggeredTransition( + BlueGreenContext context, BlueGreenDeploymentType currentType) { + + FlinkDeployment activeChild = context.getDeploymentByType(currentType); + if (activeChild == null || activeChild.getStatus() == null) { + return null; + } + + if (!isChildStableForAutoscalerPropagation(activeChild)) { + return null; + } + + var reconStatus = activeChild.getStatus().getReconciliationStatus(); + if (reconStatus == null || reconStatus.isBeforeFirstDeployment()) { + return null; + } + + FlinkDeploymentSpec childLastReconciledSpec = reconStatus.deserializeLastReconciledSpec(); + if (childLastReconciledSpec == null) { + return null; + } + + FlinkDeploymentSpec childKubeSpec = activeChild.getSpec(); + Map configOverrides = + detectConfigOverrides(childKubeSpec, childLastReconciledSpec); + String tmMemoryOverride = detectTmMemoryOverride(childKubeSpec, childLastReconciledSpec); + + if (configOverrides.isEmpty() && tmMemoryOverride == null) { + return null; // No autoscaler changes — nothing to do + } + + LOG.info( + "Autoscaler overrides detected on child '{}' (config keys: {}, TM memory: {}). " + + "Initiating Blue/Green transition.", + activeChild.getMetadata().getName(), + configOverrides.size(), + tmMemoryOverride != null); + + // Handle savepoint if required — reuses the existing savepoint flow. + // Do NOT call setLastReconciledSpec before SAVEPOINTING so overrides + // are re-detected from the child's lastReconciledSpec after completion. + try { + boolean savepointTriggered = handleSavepoint(context, activeChild); + if (savepointTriggered) { + var savepointingState = calculateSavepointingState(currentType); + return patchStatusUpdateControl(context, savepointingState, null, null) + .rescheduleAfter(getReconciliationReschedInterval(context)); + } + } catch (Exception e) { + var error = + "Could not trigger Savepoint for autoscaler transition. Details: " + + e.getMessage(); + return markDeploymentFailing(context, error, e); + } + + // Serialize the clean parent spec BEFORE applying in-memory overrides. + setLastReconciledSpec(context); + + // Capture the clean spec JSON in a local variable so the finally-block restore is + // decoupled from whatever may happen to deploymentStatus.lastReconciledSpec inside + // startTransition (defensive — nothing modifies it today, but this avoids a subtle + // implicit contract). + final String cleanSpecJson = context.getDeploymentStatus().getLastReconciledSpec(); + + // Apply overrides to the parent's in-memory spec (ephemeral — not persisted to K8s). + // prepareFlinkDeployment deep-copies from this spec, so the new child gets the overrides. + applyOverridesToParentSpec(context, configOverrides, tmMemoryOverride); + + try { + // Use the existing transition flow — same path as user-triggered transitions. + return startTransition(context, currentType, activeChild); + } catch (Exception e) { + var error = "Could not start autoscaler transition. Details: " + e.getMessage(); + context.getDeploymentStatus().setSavepointTriggerId(null); + return markDeploymentFailing(context, error, e); + } finally { + // Restore the parent's in-memory spec to the clean version (before autoscaler + // overrides). The new child already received the overrides via deep copy in + // prepareFlinkDeployment. Without this restoration, + // handleSpecChangesDuringTransition would detect a phantom diff on the next + // reconcile (parent spec with overrides vs clean lastReconciledSpec). + context.getBgDeployment() + .setSpec( + SpecUtils.readSpecFromJSON( + cleanSpecJson, "spec", FlinkBlueGreenDeploymentSpec.class)); + } + } + + /** + * Detects flinkConfiguration keys that the autoscaler added or changed in lastReconciledSpec + * relative to the child's K8s spec. Only scans keys present in lastReconciledSpec — keys + * removed from lastReconciledSpec (but still in K8s spec) are intentionally not detected, as + * autoscalers add/modify overrides but do not remove them. + */ + private Map detectConfigOverrides( + FlinkDeploymentSpec childKubeSpec, FlinkDeploymentSpec childLastReconciledSpec) { + + Map kubeConfig = + childKubeSpec.getFlinkConfiguration() != null + ? childKubeSpec.getFlinkConfiguration().asFlatMap() + : Map.of(); + Map reconConfig = + childLastReconciledSpec.getFlinkConfiguration() != null + ? childLastReconciledSpec.getFlinkConfiguration().asFlatMap() + : Map.of(); + + Map overrides = new HashMap<>(); + for (Map.Entry entry : reconConfig.entrySet()) { + if (!entry.getValue().equals(kubeConfig.get(entry.getKey()))) { + overrides.put(entry.getKey(), entry.getValue()); + } + } + return overrides; + } + + /** Detects taskManager.resource.memory override (lastReconciledSpec differs from K8s spec). */ + @Nullable + private String detectTmMemoryOverride( + FlinkDeploymentSpec childKubeSpec, FlinkDeploymentSpec childLastReconciledSpec) { + String kubeMemory = getTaskManagerMemory(childKubeSpec); + String reconMemory = getTaskManagerMemory(childLastReconciledSpec); + return (reconMemory != null && !reconMemory.equals(kubeMemory)) ? reconMemory : null; + } + + @Nullable + private String getTaskManagerMemory(FlinkDeploymentSpec spec) { + if (spec == null + || spec.getTaskManager() == null + || spec.getTaskManager().getResource() == null) { + return null; + } + return spec.getTaskManager().getResource().getMemory(); + } + + /** + * Applies autoscaler overrides to the parent's in-memory spec. This is ephemeral — the parent's + * K8s spec is never modified. {@code prepareFlinkDeployment} deep-copies from the parent spec, + * so the new child will include these overrides in its K8s spec. + */ + private void applyOverridesToParentSpec( + BlueGreenContext context, + Map configOverrides, + @Nullable String tmMemoryOverride) { + + var spec = context.getBgDeployment().getSpec().getTemplate().getSpec(); + + if (!configOverrides.isEmpty() && spec.getFlinkConfiguration() != null) { + configOverrides.forEach((k, v) -> spec.getFlinkConfiguration().put(k, v)); + } + + if (tmMemoryOverride != null) { + if (spec.getTaskManager() == null) { + spec.setTaskManager( + new org.apache.flink.kubernetes.operator.api.spec.TaskManagerSpec()); + } + if (spec.getTaskManager().getResource() == null) { + spec.getTaskManager() + .setResource(new org.apache.flink.kubernetes.operator.api.spec.Resource()); + } + spec.getTaskManager().getResource().setMemory(tmMemoryOverride); + } + } + + /** + * Checks if a child FlinkDeployment has been stable long enough to trust autoscaler changes. + * Requires STABLE lifecycle state and lastReconciledSpecStable to prevent cascading + * transitions. + */ + private boolean isChildStableForAutoscalerPropagation(FlinkDeployment child) { + if (child.getStatus() == null + || child.getStatus().getLifecycleState() != ResourceLifecycleState.STABLE) { + return false; + } + var reconStatus = child.getStatus().getReconciliationStatus(); + return reconStatus != null && reconStatus.isLastReconciledSpecStable(); + } + // ==================== Common Utility Methods ==================== public static UpdateControl patchStatusUpdateControl( diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/bluegreen/BlueGreenKubernetesService.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/bluegreen/BlueGreenKubernetesService.java index ae7492d312..2f64b1ab35 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/bluegreen/BlueGreenKubernetesService.java +++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/bluegreen/BlueGreenKubernetesService.java @@ -116,4 +116,29 @@ public static boolean deleteFlinkDeployment( return deletedStatus.size() == 1 && deletedStatus.get(0).getKind().equals("FlinkDeployment"); } + + /** + * Checks if a FlinkDeployment is owned by a FlinkBlueGreenDeployment. + * + *

This is used to determine whether in-place scaling should be skipped for this deployment, + * as Blue/Green deployments handle scaling through full transitions instead. + * + * @param deployment the FlinkDeployment to check + * @return true if the deployment is owned by a FlinkBlueGreenDeployment, false otherwise + */ + public static boolean isOwnedByBlueGreenDeployment(FlinkDeployment deployment) { + if (deployment == null || deployment.getMetadata() == null) { + return false; + } + var ownerReferences = deployment.getMetadata().getOwnerReferences(); + if (ownerReferences == null || ownerReferences.isEmpty()) { + return false; + } + return ownerReferences.stream() + .anyMatch( + ref -> + FlinkBlueGreenDeployment.class + .getSimpleName() + .equals(ref.getKind())); + } } diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java index 7f8af3ca88..87b35ad0cf 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java +++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java @@ -34,6 +34,7 @@ import org.apache.flink.kubernetes.operator.autoscaler.KubernetesJobAutoScalerContext; import org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions; import org.apache.flink.kubernetes.operator.controller.FlinkResourceContext; +import org.apache.flink.kubernetes.operator.controller.bluegreen.BlueGreenKubernetesService; import org.apache.flink.kubernetes.operator.reconciler.Reconciler; import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils; import org.apache.flink.kubernetes.operator.reconciler.diff.DiffResult; @@ -129,6 +130,22 @@ public void reconcile(FlinkResourceContext ctx) throws Exception { cr.getStatus().getReconciliationStatus().deserializeLastReconciledSpec(); SPEC currentDeploySpec = cr.getSpec(); + // For B/G owned children, capture whether there's a user/parent-initiated spec change + // BEFORE the autoscaler modifies the in-memory spec. This lets us distinguish + // autoscaler-induced changes from parent-initiated changes. + boolean isBlueGreenOwned = isBlueGreenOwnedDeployment(ctx); + boolean hasPreAutoscalerSpecChange = false; + if (isBlueGreenOwned) { + hasPreAutoscalerSpecChange = + DiffType.IGNORE + != new ReflectiveDiffBuilder<>( + ctx.getDeploymentMode(), + lastReconciledSpec, + currentDeploySpec) + .build() + .getType(); + } + applyAutoscaler(ctx); var reconciliationState = reconciliationStatus.getState(); @@ -152,6 +169,23 @@ public void reconcile(FlinkResourceContext ctx) throws Exception { if (checkNewSpecAlreadyDeployed(cr, deployConfig)) { return; } + + // For B/G owned children: if the diff is PURELY from the autoscaler + // (no pre-existing spec change from parent), skip ALL in-place processing. + // This covers both horizontal (SCALE) and vertical (UPGRADE) autoscaler changes. + // The B/G parent controller will detect this via lastReconciledSpec and trigger + // a Blue/Green transition instead. + if (isBlueGreenOwned && !hasPreAutoscalerSpecChange) { + LOG.info( + "Skipping autoscaler-induced spec change ({}) for B/G owned '{}'. " + + "Changes will be handled via Blue/Green transition.", + diffType, + cr.getMetadata().getName()); + ReconciliationUtils.updateStatusForDeployedSpec( + ctx.getResource(), deployConfig, clock); + return; + } + triggerSpecChangeEvent(cr, specDiff, ctx.getKubernetesClient()); // Try scaling if this is not an upgrade/redeploy change @@ -352,9 +386,15 @@ private boolean checkNewSpecAlreadyDeployed(CR resource, Configuration deployCon * Scale the cluster in-place if possible, either through reactive scaling or declarative * resources. * + *

Note: In-place scaling is skipped for FlinkDeployments owned by a + * FlinkBlueGreenDeployment. Blue/Green deployments handle scaling through full transitions + * instead, where autoscaler changes are propagated to the parent and trigger a new deployment. + * When scaling is skipped for B/G owned deployments, we return true to prevent the reconciler + * from attempting a full job restart, and update the status to record the spec change. + * * @param ctx Resource context. * @param deployConfig Configuration to be deployed. - * @return True if the scaling is successful + * @return True if the scaling is successful or handled externally (e.g., by B/G parent) * @throws Exception */ private boolean scale(FlinkResourceContext ctx, Configuration deployConfig) @@ -369,6 +409,18 @@ private boolean scale(FlinkResourceContext ctx, Configuration deployConfig) return scaled; } + /** + * Checks if the current resource is a FlinkDeployment owned by a FlinkBlueGreenDeployment. + * + * @param ctx the resource context + * @return true if the resource is a B/G owned FlinkDeployment + */ + private boolean isBlueGreenOwnedDeployment(FlinkResourceContext ctx) { + return ctx.getResource() instanceof FlinkDeployment + && BlueGreenKubernetesService.isOwnedByBlueGreenDeployment( + (FlinkDeployment) ctx.getResource()); + } + /** * Checks whether the currently deployed Flink resource spec should be rolled back to the stable * spec. This includes validating the current deployment status, config and checking if the last 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 af3ccbd65c..03e8377739 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 @@ -366,6 +366,20 @@ public static FlinkDeployment prepareFlinkDeployment( flinkDeployment.setSpec(spec.getTemplate().getSpec()); + // Explicitly set job.state to ensure it's not null (which could cause merge issues) + // If the parent spec has an explicit state, use it; otherwise default to RUNNING + var parentJobState = spec.getTemplate().getSpec().getJob().getState(); + if (parentJobState == null) { + flinkDeployment + .getSpec() + .getJob() + .setState(org.apache.flink.kubernetes.operator.api.spec.JobState.RUNNING); + LOG.info("Job state was null, explicitly setting to RUNNING"); + } else { + flinkDeployment.getSpec().getJob().setState(parentJobState); + LOG.info("Using explicit job state from parent: {}", parentJobState); + } + // Update Ingress template if exists to prevent path collision between Blue and Green IngressSpec ingress = flinkDeployment.getSpec().getIngress(); if (ingress != 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 528b61727d..31565d5091 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 @@ -20,6 +20,7 @@ import org.apache.flink.api.common.JobStatus; import org.apache.flink.configuration.CheckpointingOptions; import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.PipelineOptions; import org.apache.flink.configuration.TaskManagerOptions; import org.apache.flink.kubernetes.operator.TestUtils; import org.apache.flink.kubernetes.operator.TestingFlinkService; @@ -62,6 +63,7 @@ import java.util.List; import java.util.Map; import java.util.UUID; +import java.util.function.Consumer; import java.util.stream.Stream; import static org.apache.flink.kubernetes.operator.api.spec.FlinkBlueGreenDeploymentConfigOptions.ABORT_GRACE_PERIOD; @@ -257,6 +259,371 @@ public void verifySavepointRedeployNonceTriggersTransitionWithInitialSavepointPa rs.reconciledStatus.getBlueGreenState()); } + @ParameterizedTest + @MethodSource("org.apache.flink.kubernetes.operator.TestUtils#flinkVersions") + public void verifyAutoscalerChangesTriggersBlueGreenTransition(FlinkVersion flinkVersion) + throws Exception { + var rs = + executeBasicDeployment( + flinkVersion, buildAutoscalerEnabledCluster(flinkVersion), false, null); + assertEquals( + FlinkBlueGreenDeploymentState.ACTIVE_BLUE, rs.reconciledStatus.getBlueGreenState()); + + // Simulate autoscaler adding parallelism overrides to the active Blue child + String autoscalerOverrides = "vertex1:4,vertex2:8"; + simulateAutoscalerOnChild( + getChildByColor("blue"), + spec -> + spec.getFlinkConfiguration() + .put( + PipelineOptions.PARALLELISM_OVERRIDES.key(), + autoscalerOverrides)); + + // Reconcile — should trigger a B/G transition + rs = reconcile(rs.deployment); + assertEquals( + FlinkBlueGreenDeploymentState.TRANSITIONING_TO_GREEN, + rs.reconciledStatus.getBlueGreenState(), + "Autoscaler changes should trigger transition to Green"); + + // Verify: new Green child has the overrides (parent spec is NOT modified) + assertEquals( + autoscalerOverrides, + getChildByColor("green") + .getSpec() + .getFlinkConfiguration() + .asFlatMap() + .get(PipelineOptions.PARALLELISM_OVERRIDES.key()), + "Green deployment should have the parallelism overrides"); + + // Complete the transition + rs = completeTransition(rs, "green"); + assertEquals( + FlinkBlueGreenDeploymentState.ACTIVE_GREEN, + rs.reconciledStatus.getBlueGreenState(), + "Should be ACTIVE_GREEN after transition completes"); + } + + @ParameterizedTest + @MethodSource("org.apache.flink.kubernetes.operator.TestUtils#flinkVersions") + public void verifyVerticalAutoscalerChangesTriggerBlueGreenTransition(FlinkVersion flinkVersion) + throws Exception { + // Setup: Deploy with initial TM memory and autoscaler enabled + var deployment = buildAutoscalerEnabledCluster(flinkVersion); + var tmSpec = new TaskManagerSpec(); + var tmResource = new Resource(); + tmResource.setMemory("2048m"); + tmSpec.setResource(tmResource); + deployment.getSpec().getTemplate().getSpec().setTaskManager(tmSpec); + + var rs = executeBasicDeployment(flinkVersion, deployment, false, null); + + // Simulate autoscaler performing vertical scaling (memory tuning) + String scaledMemory = String.valueOf(4096 * 1024 * 1024L); + simulateAutoscalerOnChild( + getChildByColor("blue"), + spec -> { + spec.getFlinkConfiguration() + .put(TaskManagerOptions.TOTAL_PROCESS_MEMORY.key(), scaledMemory); + spec.getFlinkConfiguration() + .put(TaskManagerOptions.FRAMEWORK_HEAP_MEMORY.key(), "0 bytes"); + spec.getTaskManager().getResource().setMemory(scaledMemory); + }); + + // Reconcile — should trigger a B/G transition + rs = reconcile(rs.deployment); + assertEquals( + FlinkBlueGreenDeploymentState.TRANSITIONING_TO_GREEN, + rs.reconciledStatus.getBlueGreenState(), + "Vertical autoscaler changes should trigger transition to Green"); + + // Verify Green child has the updated memory + var greenChild = getChildByColor("green"); + assertEquals( + scaledMemory, + greenChild.getSpec().getTaskManager().getResource().getMemory(), + "Green deployment should have the autoscaler's TM memory"); + assertNotNull( + greenChild + .getSpec() + .getFlinkConfiguration() + .get(TaskManagerOptions.TOTAL_PROCESS_MEMORY.key()), + "Green deployment should have TM total process memory config"); + } + + @ParameterizedTest + @MethodSource("org.apache.flink.kubernetes.operator.TestUtils#flinkVersions") + public void verifyUserParallelismChangeAfterAutoscalerTriggersTransition( + FlinkVersion flinkVersion) throws Exception { + var rs = + executeBasicDeployment( + flinkVersion, buildAutoscalerEnabledCluster(flinkVersion), false, null); + + // Autoscaler adds parallelism overrides → B/G transition → complete to ACTIVE_GREEN + simulateAutoscalerOnChild( + getChildByColor("blue"), + spec -> + spec.getFlinkConfiguration() + .put( + PipelineOptions.PARALLELISM_OVERRIDES.key(), + "vertex1:4,vertex2:8")); + rs = reconcile(rs.deployment); + assertEquals( + FlinkBlueGreenDeploymentState.TRANSITIONING_TO_GREEN, + rs.reconciledStatus.getBlueGreenState()); + rs = completeTransition(rs, "green"); + assertEquals( + FlinkBlueGreenDeploymentState.ACTIVE_GREEN, + rs.reconciledStatus.getBlueGreenState()); + + // User manually sets parallelism overrides → should trigger a new B/G transition. + rs.deployment = kubernetesClient.resource(rs.deployment).get(); + rs.deployment + .getSpec() + .getTemplate() + .getSpec() + .getFlinkConfiguration() + .put(PipelineOptions.PARALLELISM_OVERRIDES.key(), "vertex1:2,vertex2:4"); + kubernetesClient.resource(rs.deployment).createOrReplace(); + + rs = reconcile(rs.deployment); + assertEquals( + FlinkBlueGreenDeploymentState.TRANSITIONING_TO_BLUE, + rs.reconciledStatus.getBlueGreenState(), + "User parallelism change after autoscaler should trigger a new transition"); + } + + @ParameterizedTest + @MethodSource("org.apache.flink.kubernetes.operator.TestUtils#flinkVersions") + public void verifyUserTmMemoryChangeAfterAutoscalerTriggersTransition(FlinkVersion flinkVersion) + throws Exception { + var deployment = buildAutoscalerEnabledCluster(flinkVersion); + deployment + .getSpec() + .getTemplate() + .getSpec() + .getFlinkConfiguration() + .put("job.autoscaler.memory.tuning.enabled", "true"); + var tmSpec = new TaskManagerSpec(); + var tmResource = new Resource(); + tmResource.setMemory("2048m"); + tmSpec.setResource(tmResource); + deployment.getSpec().getTemplate().getSpec().setTaskManager(tmSpec); + + var rs = executeBasicDeployment(flinkVersion, deployment, false, null); + + // Autoscaler changes TM memory → B/G transition → complete to ACTIVE_GREEN + simulateAutoscalerOnChild( + getChildByColor("blue"), + spec -> spec.getTaskManager().getResource().setMemory("4096m")); + rs = reconcile(rs.deployment); + assertEquals( + FlinkBlueGreenDeploymentState.TRANSITIONING_TO_GREEN, + rs.reconciledStatus.getBlueGreenState()); + rs = completeTransition(rs, "green"); + assertEquals( + FlinkBlueGreenDeploymentState.ACTIVE_GREEN, + rs.reconciledStatus.getBlueGreenState()); + + // User manually changes TM memory → should trigger a new B/G transition. + rs.deployment = kubernetesClient.resource(rs.deployment).get(); + rs.deployment + .getSpec() + .getTemplate() + .getSpec() + .getTaskManager() + .getResource() + .setMemory("3072m"); + kubernetesClient.resource(rs.deployment).createOrReplace(); + + rs = reconcile(rs.deployment); + assertEquals( + FlinkBlueGreenDeploymentState.TRANSITIONING_TO_BLUE, + rs.reconciledStatus.getBlueGreenState(), + "User TM memory change after autoscaler should trigger a new transition"); + } + + @ParameterizedTest + @MethodSource("org.apache.flink.kubernetes.operator.TestUtils#flinkVersions") + public void verifyAutoscalerPhantomDiffPreventionAfterTransition(FlinkVersion flinkVersion) + throws Exception { + var rs = + executeBasicDeployment( + flinkVersion, buildAutoscalerEnabledCluster(flinkVersion), false, null); + + // Autoscaler adds parallelism overrides → B/G transition → complete + simulateAutoscalerOnChild( + getChildByColor("blue"), + spec -> + spec.getFlinkConfiguration() + .put( + PipelineOptions.PARALLELISM_OVERRIDES.key(), + "vertex1:4,vertex2:8")); + rs = reconcile(rs.deployment); + assertEquals( + FlinkBlueGreenDeploymentState.TRANSITIONING_TO_GREEN, + rs.reconciledStatus.getBlueGreenState()); + rs = completeTransition(rs, "green"); + assertEquals( + FlinkBlueGreenDeploymentState.ACTIVE_GREEN, + rs.reconciledStatus.getBlueGreenState()); + + // Subsequent reconciles should NOT trigger spurious transitions. + // Parent spec and lastReconciledSpec were never modified with autoscaler values, + // so there is no phantom diff. The green child was created with overrides baked into + // its K8s spec, and its lastReconciledSpec matches (no further autoscaler diff). + rs = reconcile(rs.deployment); + assertEquals( + FlinkBlueGreenDeploymentState.ACTIVE_GREEN, + rs.reconciledStatus.getBlueGreenState(), + "No phantom diff — should remain ACTIVE_GREEN"); + rs = reconcile(rs.deployment); + assertEquals( + FlinkBlueGreenDeploymentState.ACTIVE_GREEN, + rs.reconciledStatus.getBlueGreenState(), + "Should still be ACTIVE_GREEN after multiple reconciles"); + } + + @ParameterizedTest + @MethodSource("org.apache.flink.kubernetes.operator.TestUtils#flinkVersions") + public void verifyAutoscalerRetriggersAfterAbortedTransition(FlinkVersion flinkVersion) + throws Exception { + var rs = + executeBasicDeployment( + flinkVersion, buildAutoscalerEnabledCluster(flinkVersion), false, null); + + // Autoscaler adds parallelism overrides → triggers B/G transition + String autoscalerOverrides = "vertex1:4,vertex2:8"; + simulateAutoscalerOnChild( + getChildByColor("blue"), + spec -> + spec.getFlinkConfiguration() + .put( + PipelineOptions.PARALLELISM_OVERRIDES.key(), + autoscalerOverrides)); + rs = reconcile(rs.deployment); + assertEquals( + FlinkBlueGreenDeploymentState.TRANSITIONING_TO_GREEN, + rs.reconciledStatus.getBlueGreenState()); + + // Green child fails to start — abort after grace period. + Thread.sleep(MINIMUM_ABORT_GRACE_PERIOD); + rs = reconcile(rs.deployment); + assertEquals( + FlinkBlueGreenDeploymentState.ACTIVE_BLUE, + rs.reconciledStatus.getBlueGreenState(), + "Should roll back to ACTIVE_BLUE after abort"); + + // Clean up stale Green child. Blue is already stable with autoscaler overrides + // still in its lastReconciledSpec (abort is a parent-level operation and does not + // touch the Blue child's status). + try { + kubernetesClient.resource(getChildByColor("green")).delete(); + } catch (AssertionError ignored) { + // Green may already be deleted + } + + // Next reconcile should re-detect autoscaler diff and re-trigger transition + rs = reconcile(rs.deployment); + assertEquals( + FlinkBlueGreenDeploymentState.TRANSITIONING_TO_GREEN, + rs.reconciledStatus.getBlueGreenState(), + "Should re-trigger transition after abort (not permanently stuck)"); + } + + @ParameterizedTest + @MethodSource("org.apache.flink.kubernetes.operator.TestUtils#flinkVersions") + public void verifyAutoscalerTransitionWithSavepointUpgradeMode(FlinkVersion flinkVersion) + throws Exception { + // Setup: SAVEPOINT upgrade mode with autoscaler enabled + var deployment = + buildSessionCluster( + TEST_DEPLOYMENT_NAME, + TEST_NAMESPACE, + flinkVersion, + null, + UpgradeMode.SAVEPOINT); + deployment + .getSpec() + .getTemplate() + .getSpec() + .getFlinkConfiguration() + .put("job.autoscaler.enabled", "true"); + + var rs = executeBasicDeployment(flinkVersion, deployment, false, null); + assertEquals( + FlinkBlueGreenDeploymentState.ACTIVE_BLUE, rs.reconciledStatus.getBlueGreenState()); + + // Simulate autoscaler adding parallelism overrides to the active Blue child + String autoscalerOverrides = "vertex1:4,vertex2:8"; + simulateAutoscalerOnChild( + getChildByColor("blue"), + spec -> + spec.getFlinkConfiguration() + .put( + PipelineOptions.PARALLELISM_OVERRIDES.key(), + autoscalerOverrides)); + + // Drive the savepoint flow — reuse the existing handleSavepoint helper. + // First reconcile detects overrides and triggers a savepoint → SAVEPOINTING_BLUE + var triggers = flinkService.getSavepointTriggers(); + triggers.clear(); + + rs = reconcile(rs.deployment); + + // Simulating a pending savepoint + triggers.put(rs.deployment.getStatus().getSavepointTriggerId(), false); + + assertEquals( + FlinkBlueGreenDeploymentState.SAVEPOINTING_BLUE, + rs.reconciledStatus.getBlueGreenState(), + "Autoscaler transition should enter SAVEPOINTING_BLUE for SAVEPOINT upgrade mode"); + assertTrue(rs.updateControl.isPatchStatus()); + + // Next reconciliation waits on the pending savepoint + rs = reconcile(rs.deployment); + assertTrue(rs.updateControl.isNoUpdate()); + + // Complete the savepoint + triggers.put(rs.deployment.getStatus().getSavepointTriggerId(), true); + rs = reconcile(rs.deployment); + + // Should return to ACTIVE_BLUE after savepoint completes + assertEquals( + FlinkBlueGreenDeploymentState.ACTIVE_BLUE, + rs.reconciledStatus.getBlueGreenState(), + "Should return to ACTIVE_BLUE after savepoint completes"); + + // Next reconcile re-detects autoscaler overrides and starts the actual transition + rs = reconcile(rs.deployment); + assertEquals( + FlinkBlueGreenDeploymentState.TRANSITIONING_TO_GREEN, + rs.reconciledStatus.getBlueGreenState(), + "Should transition to Green after savepoint completes"); + + // Verify: Green child has the autoscaler overrides + assertEquals( + autoscalerOverrides, + getChildByColor("green") + .getSpec() + .getFlinkConfiguration() + .asFlatMap() + .get(PipelineOptions.PARALLELISM_OVERRIDES.key()), + "Green deployment should have the parallelism overrides"); + + // Verify: Green child uses the savepoint (not null) + assertNotNull( + getChildByColor("green").getSpec().getJob().getInitialSavepointPath(), + "Green deployment should use a savepoint path"); + + // Complete the transition + rs = completeTransition(rs, "green"); + assertEquals( + FlinkBlueGreenDeploymentState.ACTIVE_GREEN, + rs.reconciledStatus.getBlueGreenState(), + "Should be ACTIVE_GREEN after transition completes"); + } + @ParameterizedTest @MethodSource("org.apache.flink.kubernetes.operator.TestUtils#flinkVersions") public void verifySuspendAndResumeInPlace(FlinkVersion flinkVersion) throws Exception { @@ -1107,6 +1474,64 @@ static class ReconcileResult { flinkVersion, blueGreenDeployment, false, TEST_INITIAL_SAVEPOINT_PATH); } + // ---- Autoscaler Test Helpers ---- + + /** Builds a session cluster with the autoscaler enabled in the parent's flinkConfiguration. */ + private FlinkBlueGreenDeployment buildAutoscalerEnabledCluster(FlinkVersion flinkVersion) { + var deployment = + buildSessionCluster( + TEST_DEPLOYMENT_NAME, + TEST_NAMESPACE, + flinkVersion, + null, + UpgradeMode.STATELESS); + deployment + .getSpec() + .getTemplate() + .getSpec() + .getFlinkConfiguration() + .put("job.autoscaler.enabled", "true"); + return deployment; + } + + /** + * Simulates the autoscaler modifying the child's in-memory spec. Applies the given mutation to + * the child's spec, serializes it into lastReconciledSpec, marks it stable, and updates status. + */ + private void simulateAutoscalerOnChild( + FlinkDeployment child, Consumer specMutation) { + var spec = child.getSpec(); + specMutation.accept(spec); + child.getStatus().getReconciliationStatus().serializeAndSetLastReconciledSpec(spec, child); + child.getStatus().getReconciliationStatus().markReconciledSpecAsStable(); + kubernetesClient.resource(child).updateStatus(); + } + + /** Finds a child FlinkDeployment by color suffix ("blue" or "green"). */ + private FlinkDeployment getChildByColor(String color) { + return getFlinkDeployments().stream() + .filter(d -> d.getMetadata().getName().endsWith("-" + color)) + .findFirst() + .orElseThrow(() -> new AssertionError("No " + color + " child found")); + } + + /** + * Drives a B/G transition to completion: simulates a successful job start on the target child, + * waits for the deletion delay, and reconciles until the new color is active. + */ + private TestingFlinkBlueGreenDeploymentController.BlueGreenReconciliationResult + completeTransition( + TestingFlinkBlueGreenDeploymentController.BlueGreenReconciliationResult rs, + String targetColor) + throws Exception { + simulateSuccessfulJobStart(getChildByColor(targetColor)); + rs = reconcile(rs.deployment); + Thread.sleep(rs.updateControl.getScheduleDelay().get()); + reconcile(rs.deployment); + assertEquals(1, getFlinkDeployments().size()); + return reconcile(rs.deployment); + } + private ReconcileResult reconcileAndVerifyPatchBehavior( TestingFlinkBlueGreenDeploymentController.BlueGreenReconciliationResult rs) throws Exception { From ec888f6a49d1ea956934d22f1a6dbc2af3829c36 Mon Sep 17 00:00:00 2001 From: James Kan Date: Thu, 12 Feb 2026 11:24:41 -0800 Subject: [PATCH 2/5] Improve robustness and reuse existing logics --- .../bluegreen/BlueGreenDeploymentService.java | 214 +++++++++--------- .../AbstractFlinkResourceReconciler.java | 8 +- .../utils/bluegreen/BlueGreenUtils.java | 54 +++-- ...linkBlueGreenDeploymentControllerTest.java | 15 +- 4 files changed, 145 insertions(+), 146 deletions(-) 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 a5e296b56d..06f1915d92 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,9 +22,12 @@ 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.diff.DiffType; import org.apache.flink.kubernetes.operator.api.lifecycle.ResourceLifecycleState; 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 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; @@ -33,6 +36,7 @@ import org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions; import org.apache.flink.kubernetes.operator.controller.FlinkBlueGreenDeployments; import org.apache.flink.kubernetes.operator.controller.FlinkResourceContext; +import org.apache.flink.kubernetes.operator.reconciler.diff.ReflectiveDiffBuilder; import org.apache.flink.kubernetes.operator.utils.IngressUtils; import org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenUtils; import org.apache.flink.util.Preconditions; @@ -47,8 +51,6 @@ import org.slf4j.LoggerFactory; import java.time.Instant; -import java.util.HashMap; -import java.util.Map; import java.util.Objects; import static org.apache.flink.kubernetes.operator.controller.bluegreen.BlueGreenKubernetesService.deleteFlinkDeployment; @@ -89,6 +91,29 @@ public UpdateControl initiateDeployment( FlinkBlueGreenDeploymentState nextState, Savepoint lastCheckpoint, boolean isFirstDeployment) { + return initiateDeployment( + context, + nextBlueGreenDeploymentType, + nextState, + lastCheckpoint, + isFirstDeployment, + null); + } + + /** + * Initiates a new Blue/Green deployment, optionally using a pre-built spec override. + * + * @param specOverride if non-null, used as the source spec for child creation instead of the + * parent's current spec. This is used by autoscaler-triggered transitions to pass overrides + * without mutating the parent's in-memory spec. + */ + private UpdateControl initiateDeployment( + BlueGreenContext context, + BlueGreenDeploymentType nextBlueGreenDeploymentType, + FlinkBlueGreenDeploymentState nextState, + Savepoint lastCheckpoint, + boolean isFirstDeployment, + @Nullable FlinkBlueGreenDeploymentSpec specOverride) { ObjectMeta bgMeta = context.getBgDeployment().getMetadata(); FlinkDeployment flinkDeployment = @@ -97,7 +122,8 @@ public UpdateControl initiateDeployment( nextBlueGreenDeploymentType, lastCheckpoint, isFirstDeployment, - bgMeta); + bgMeta, + specOverride); deployCluster(context, flinkDeployment); @@ -243,9 +269,7 @@ private boolean isChildSuspended(FlinkDeployment deployment) { return false; } var job = deployment.getSpec().getJob(); - return job != null - && job.getState() - == org.apache.flink.kubernetes.operator.api.spec.JobState.SUSPENDED; + return job != null && job.getState() == JobState.SUSPENDED; } private UpdateControl patchFlinkDeployment( @@ -324,6 +348,21 @@ private UpdateControl startTransition( BlueGreenContext context, BlueGreenDeploymentType currentBlueGreenDeploymentType, FlinkDeployment currentFlinkDeployment) { + return startTransition( + context, currentBlueGreenDeploymentType, currentFlinkDeployment, null); + } + + /** + * Starts a B/G transition, optionally using a pre-built spec override for the new child. + * + * @param specOverride if non-null, passed through to {@code prepareFlinkDeployment} so the new + * child is created from this spec instead of the parent's current spec. + */ + private UpdateControl startTransition( + BlueGreenContext context, + BlueGreenDeploymentType currentBlueGreenDeploymentType, + FlinkDeployment currentFlinkDeployment, + @Nullable FlinkBlueGreenDeploymentSpec specOverride) { DeploymentTransition transition = calculateTransition(currentBlueGreenDeploymentType); LOG.info( @@ -340,7 +379,8 @@ private UpdateControl startTransition( transition.nextBlueGreenDeploymentType, transition.nextState, lastCheckpoint, - false); + false, + specOverride); } /** @@ -854,9 +894,9 @@ public void updateBlueGreenIngress( /** * Detects autoscaler overrides on the active child and initiates a B/G transition if any are - * found. Overrides are applied ephemerally to the parent's in-memory spec so the new child - * inherits them via {@code prepareFlinkDeployment}; the parent spec is restored in a finally - * block to prevent phantom diffs. + * found. A merged spec copy (parent spec + overrides) is built and passed explicitly to the + * transition flow so the new child inherits the overrides without mutating the parent's + * in-memory spec. * * @return UpdateControl if a transition was initiated, or {@code null} if no overrides */ @@ -884,20 +924,26 @@ private UpdateControl checkAutoscalerTriggeredTransiti } FlinkDeploymentSpec childKubeSpec = activeChild.getSpec(); - Map configOverrides = - detectConfigOverrides(childKubeSpec, childLastReconciledSpec); - String tmMemoryOverride = detectTmMemoryOverride(childKubeSpec, childLastReconciledSpec); - if (configOverrides.isEmpty() && tmMemoryOverride == null) { + // Use the same diff infrastructure as the child reconciler to detect overrides. + // This keeps both layers in sync — any field the autoscaler modifies is caught + // automatically, including future autoscaler capabilities. + var autoscalerDiff = + new ReflectiveDiffBuilder<>( + KubernetesDeploymentMode.NATIVE, + childKubeSpec, + childLastReconciledSpec) + .build(); + + if (autoscalerDiff.getType() == DiffType.IGNORE) { return null; // No autoscaler changes — nothing to do } LOG.info( - "Autoscaler overrides detected on child '{}' (config keys: {}, TM memory: {}). " + "Autoscaler overrides detected on child '{}' (diffType: {}). " + "Initiating Blue/Green transition.", activeChild.getMetadata().getName(), - configOverrides.size(), - tmMemoryOverride != null); + autoscalerDiff.getType()); // Handle savepoint if required — reuses the existing savepoint flow. // Do NOT call setLastReconciledSpec before SAVEPOINTING so overrides @@ -916,112 +962,58 @@ private UpdateControl checkAutoscalerTriggeredTransiti return markDeploymentFailing(context, error, e); } - // Serialize the clean parent spec BEFORE applying in-memory overrides. + // Serialize the clean parent spec (without overrides) as the lastReconciledSpec. setLastReconciledSpec(context); - // Capture the clean spec JSON in a local variable so the finally-block restore is - // decoupled from whatever may happen to deploymentStatus.lastReconciledSpec inside - // startTransition (defensive — nothing modifies it today, but this avoids a subtle - // implicit contract). - final String cleanSpecJson = context.getDeploymentStatus().getLastReconciledSpec(); - - // Apply overrides to the parent's in-memory spec (ephemeral — not persisted to K8s). - // prepareFlinkDeployment deep-copies from this spec, so the new child gets the overrides. - applyOverridesToParentSpec(context, configOverrides, tmMemoryOverride); + // Build a merged spec: deep-copy the parent and swap in the child's + // lastReconciledSpec as the template. Since we only reach this path when + // there are no user spec changes, the child's lastReconciledSpec already + // equals the parent's template spec + all autoscaler modifications. + FlinkBlueGreenDeploymentSpec mergedSpec = buildMergedSpec(context, childLastReconciledSpec); try { - // Use the existing transition flow — same path as user-triggered transitions. - return startTransition(context, currentType, activeChild); + return startTransition(context, currentType, activeChild, mergedSpec); } catch (Exception e) { var error = "Could not start autoscaler transition. Details: " + e.getMessage(); context.getDeploymentStatus().setSavepointTriggerId(null); return markDeploymentFailing(context, error, e); - } finally { - // Restore the parent's in-memory spec to the clean version (before autoscaler - // overrides). The new child already received the overrides via deep copy in - // prepareFlinkDeployment. Without this restoration, - // handleSpecChangesDuringTransition would detect a phantom diff on the next - // reconcile (parent spec with overrides vs clean lastReconciledSpec). - context.getBgDeployment() - .setSpec( - SpecUtils.readSpecFromJSON( - cleanSpecJson, "spec", FlinkBlueGreenDeploymentSpec.class)); - } - } - - /** - * Detects flinkConfiguration keys that the autoscaler added or changed in lastReconciledSpec - * relative to the child's K8s spec. Only scans keys present in lastReconciledSpec — keys - * removed from lastReconciledSpec (but still in K8s spec) are intentionally not detected, as - * autoscalers add/modify overrides but do not remove them. - */ - private Map detectConfigOverrides( - FlinkDeploymentSpec childKubeSpec, FlinkDeploymentSpec childLastReconciledSpec) { - - Map kubeConfig = - childKubeSpec.getFlinkConfiguration() != null - ? childKubeSpec.getFlinkConfiguration().asFlatMap() - : Map.of(); - Map reconConfig = - childLastReconciledSpec.getFlinkConfiguration() != null - ? childLastReconciledSpec.getFlinkConfiguration().asFlatMap() - : Map.of(); - - Map overrides = new HashMap<>(); - for (Map.Entry entry : reconConfig.entrySet()) { - if (!entry.getValue().equals(kubeConfig.get(entry.getKey()))) { - overrides.put(entry.getKey(), entry.getValue()); - } } - return overrides; - } - - /** Detects taskManager.resource.memory override (lastReconciledSpec differs from K8s spec). */ - @Nullable - private String detectTmMemoryOverride( - FlinkDeploymentSpec childKubeSpec, FlinkDeploymentSpec childLastReconciledSpec) { - String kubeMemory = getTaskManagerMemory(childKubeSpec); - String reconMemory = getTaskManagerMemory(childLastReconciledSpec); - return (reconMemory != null && !reconMemory.equals(kubeMemory)) ? reconMemory : null; - } - - @Nullable - private String getTaskManagerMemory(FlinkDeploymentSpec spec) { - if (spec == null - || spec.getTaskManager() == null - || spec.getTaskManager().getResource() == null) { - return null; - } - return spec.getTaskManager().getResource().getMemory(); } /** - * Applies autoscaler overrides to the parent's in-memory spec. This is ephemeral — the parent's - * K8s spec is never modified. {@code prepareFlinkDeployment} deep-copies from the parent spec, - * so the new child will include these overrides in its K8s spec. + * Builds a spec for the new child by deep-copying the parent and overwriting only the fields + * the autoscaler modifies ({@code flinkConfiguration} and {@code taskManager}). + * + *

We copy these fields wholesale from the child's {@code lastReconciledSpec} rather + * than computing per-key deltas. Because we only reach this path when there are no user spec + * changes, the child's config/TM equals {@code parentValue + autoscalerDelta}; copying the + * whole field implicitly handles additions, changes, and removals. + * + *

We intentionally do not replace the entire {@code template.spec} because the + * child's {@link FlinkDeploymentSpec} contains deployment-specific transformations applied by + * {@code prepareFlinkDeployment} (e.g., ingress template prefixed with {@code blue-}/{@code + * green-}) that must not leak back into the parent's template. */ - private void applyOverridesToParentSpec( - BlueGreenContext context, - Map configOverrides, - @Nullable String tmMemoryOverride) { - - var spec = context.getBgDeployment().getSpec().getTemplate().getSpec(); - - if (!configOverrides.isEmpty() && spec.getFlinkConfiguration() != null) { - configOverrides.forEach((k, v) -> spec.getFlinkConfiguration().put(k, v)); - } - - if (tmMemoryOverride != null) { - if (spec.getTaskManager() == null) { - spec.setTaskManager( - new org.apache.flink.kubernetes.operator.api.spec.TaskManagerSpec()); - } - if (spec.getTaskManager().getResource() == null) { - spec.getTaskManager() - .setResource(new org.apache.flink.kubernetes.operator.api.spec.Resource()); - } - spec.getTaskManager().getResource().setMemory(tmMemoryOverride); - } + private FlinkBlueGreenDeploymentSpec buildMergedSpec( + BlueGreenContext context, FlinkDeploymentSpec childLastReconciledSpec) { + + // Deep copy the parent spec via JSON round-trip to avoid mutation. + FlinkBlueGreenDeploymentSpec merged = + SpecUtils.readSpecFromJSON( + SpecUtils.writeSpecAsJSON(context.getBgDeployment().getSpec(), "spec"), + "spec", + FlinkBlueGreenDeploymentSpec.class); + + // Overwrite only the autoscaler-managed fields (KubernetesScalingRealizer touches + // flinkConfiguration and taskManager.resource.memory — nothing else). + // Copying wholesale avoids per-key delta logic and naturally propagates config + // key removals (e.g., from memory tuning) that would otherwise cause phantom + // transition loops. + var parentFlinkSpec = merged.getTemplate().getSpec(); + parentFlinkSpec.setFlinkConfiguration(childLastReconciledSpec.getFlinkConfiguration()); + parentFlinkSpec.setTaskManager(childLastReconciledSpec.getTaskManager()); + + return merged; } /** @@ -1065,7 +1057,7 @@ public static UpdateControl patchStatusUpdateControl( deploymentStatus.setError(null); } - deploymentStatus.setLastReconciledTimestamp(java.time.Instant.now().toString()); + deploymentStatus.setLastReconciledTimestamp(Instant.now().toString()); flinkBlueGreenDeployment.setStatus(deploymentStatus); return UpdateControl.patchStatus(flinkBlueGreenDeployment); } diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java index 87b35ad0cf..3b08f3f36f 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java +++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java @@ -386,15 +386,9 @@ private boolean checkNewSpecAlreadyDeployed(CR resource, Configuration deployCon * Scale the cluster in-place if possible, either through reactive scaling or declarative * resources. * - *

Note: In-place scaling is skipped for FlinkDeployments owned by a - * FlinkBlueGreenDeployment. Blue/Green deployments handle scaling through full transitions - * instead, where autoscaler changes are propagated to the parent and trigger a new deployment. - * When scaling is skipped for B/G owned deployments, we return true to prevent the reconciler - * from attempting a full job restart, and update the status to record the spec change. - * * @param ctx Resource context. * @param deployConfig Configuration to be deployed. - * @return True if the scaling is successful or handled externally (e.g., by B/G parent) + * @return True if the scaling is successful * @throws Exception */ private boolean scale(FlinkResourceContext ctx, Configuration deployConfig) 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 03e8377739..b17fbc22d0 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 @@ -27,6 +27,7 @@ import org.apache.flink.kubernetes.operator.api.bluegreen.BlueGreenDiffType; import org.apache.flink.kubernetes.operator.api.spec.FlinkBlueGreenDeploymentSpec; import org.apache.flink.kubernetes.operator.api.spec.IngressSpec; +import org.apache.flink.kubernetes.operator.api.spec.JobState; import org.apache.flink.kubernetes.operator.api.spec.KubernetesDeploymentMode; import org.apache.flink.kubernetes.operator.api.spec.UpgradeMode; import org.apache.flink.kubernetes.operator.api.status.FlinkBlueGreenDeploymentStatus; @@ -40,6 +41,7 @@ import org.apache.flink.util.Preconditions; import io.fabric8.kubernetes.api.model.ObjectMeta; +import org.jetbrains.annotations.Nullable; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -179,9 +181,7 @@ public static long getDeploymentDeletionDelay(BlueGreenContext context) { * @return abort grace period in milliseconds */ public static long getAbortGracePeriod(BlueGreenContext context) { - long abortGracePeriod = - getConfigOption(context.getBgDeployment(), ABORT_GRACE_PERIOD).toMillis(); - return abortGracePeriod; + return getConfigOption(context.getBgDeployment(), ABORT_GRACE_PERIOD).toMillis(); } /** @@ -213,9 +213,7 @@ public static boolean isSavepointRequired(BlueGreenContext context) { .getSpec() .getJob() .getUpgradeMode(); - // return UpgradeMode.SAVEPOINT == upgradeMode; // Currently taking savepoints for all modes except STATELESS - // (previously only SAVEPOINT mode required savepoints) return UpgradeMode.STATELESS != upgradeMode; } @@ -319,9 +317,28 @@ public static FlinkDeployment prepareFlinkDeployment( Savepoint lastCheckpoint, boolean isFirstDeployment, ObjectMeta bgMeta) { + return prepareFlinkDeployment( + context, blueGreenDeploymentType, lastCheckpoint, isFirstDeployment, bgMeta, null); + } + + /** + * Creates a new FlinkDeployment resource for a Blue/Green deployment transition, optionally + * using a pre-built spec override (e.g., with autoscaler overrides already applied). + * + * @param specOverride if non-null, used as the source spec instead of the parent's current + * spec. This avoids mutating the parent's in-memory spec for ephemeral overrides. + */ + public static FlinkDeployment prepareFlinkDeployment( + BlueGreenContext context, + BlueGreenDeploymentType blueGreenDeploymentType, + Savepoint lastCheckpoint, + boolean isFirstDeployment, + ObjectMeta bgMeta, + @Nullable FlinkBlueGreenDeploymentSpec specOverride) { // Deployment FlinkDeployment flinkDeployment = new FlinkDeployment(); - FlinkBlueGreenDeploymentSpec originalSpec = context.getBgDeployment().getSpec(); + FlinkBlueGreenDeploymentSpec originalSpec = + specOverride != null ? specOverride : context.getBgDeployment().getSpec(); String childDeploymentName = bgMeta.getName() + "-" + blueGreenDeploymentType.toString().toLowerCase(); @@ -343,14 +360,13 @@ public static FlinkDeployment prepareFlinkDeployment( String initialSavepointPath = spec.getTemplate().getSpec().getJob().getInitialSavepointPath(); if (initialSavepointPath != null && !initialSavepointPath.isEmpty()) { - LOG.info("Using initialSavepointPath: " + initialSavepointPath); - spec.getTemplate().getSpec().getJob().setInitialSavepointPath(initialSavepointPath); + LOG.info("Using initialSavepointPath: {}", initialSavepointPath); } else { LOG.info("Clean startup with no checkpoint/savepoint restoration"); } } else if (lastCheckpoint != null) { String location = lastCheckpoint.getLocation().replace("file:", ""); - LOG.info("Using Blue/Green savepoint/checkpoint: " + location); + LOG.info("Using Blue/Green savepoint/checkpoint: {}", location); spec.getTemplate().getSpec().getJob().setInitialSavepointPath(location); } else { String initialSavepointPath = @@ -366,18 +382,14 @@ public static FlinkDeployment prepareFlinkDeployment( flinkDeployment.setSpec(spec.getTemplate().getSpec()); - // Explicitly set job.state to ensure it's not null (which could cause merge issues) - // If the parent spec has an explicit state, use it; otherwise default to RUNNING - var parentJobState = spec.getTemplate().getSpec().getJob().getState(); - if (parentJobState == null) { - flinkDeployment - .getSpec() - .getJob() - .setState(org.apache.flink.kubernetes.operator.api.spec.JobState.RUNNING); - LOG.info("Job state was null, explicitly setting to RUNNING"); - } else { - flinkDeployment.getSpec().getJob().setState(parentJobState); - LOG.info("Using explicit job state from parent: {}", parentJobState); + // Ensure job.state is never null (which could cause merge issues downstream). + // The spec was already copied on the line above, so this only needs to handle the + // null-state defensive case. + if (flinkDeployment.getSpec().getJob().getState() == null) { + flinkDeployment.getSpec().getJob().setState(JobState.RUNNING); + LOG.warn( + "Job state was null on parent spec for '{}', defaulting to RUNNING", + flinkDeployment.getMetadata().getName()); } // Update Ingress template if exists to prevent path collision between Blue and Green 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 31565d5091..c462dfdd51 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 @@ -486,7 +486,7 @@ public void verifyAutoscalerPhantomDiffPreventionAfterTransition(FlinkVersion fl @ParameterizedTest @MethodSource("org.apache.flink.kubernetes.operator.TestUtils#flinkVersions") - public void verifyAutoscalerRetriggersAfterAbortedTransition(FlinkVersion flinkVersion) + public void verifyAutoscalerDoesNotRetriggerAfterAbortedTransition(FlinkVersion flinkVersion) throws Exception { var rs = executeBasicDeployment( @@ -514,21 +514,22 @@ public void verifyAutoscalerRetriggersAfterAbortedTransition(FlinkVersion flinkV rs.reconciledStatus.getBlueGreenState(), "Should roll back to ACTIVE_BLUE after abort"); - // Clean up stale Green child. Blue is already stable with autoscaler overrides - // still in its lastReconciledSpec (abort is a parent-level operation and does not - // touch the Blue child's status). + // Clean up stale Green child. try { kubernetesClient.resource(getChildByColor("green")).delete(); } catch (AssertionError ignored) { // Green may already be deleted } - // Next reconcile should re-detect autoscaler diff and re-trigger transition + // After abort the parent is in FAILING state. The FAILING guard prevents the + // autoscaler from re-triggering the same overrides that just failed — this avoids + // an infinite abort→re-trigger→abort loop. A user spec change is required to + // clear FAILING and unblock new transitions. rs = reconcile(rs.deployment); assertEquals( - FlinkBlueGreenDeploymentState.TRANSITIONING_TO_GREEN, + FlinkBlueGreenDeploymentState.ACTIVE_BLUE, rs.reconciledStatus.getBlueGreenState(), - "Should re-trigger transition after abort (not permanently stuck)"); + "Should remain ACTIVE_BLUE (FAILING) — autoscaler must not re-trigger after abort"); } @ParameterizedTest From 2e5421db758b53879b97bd2fd29512fe7bdef5ce Mon Sep 17 00:00:00 2001 From: James Kan Date: Thu, 12 Feb 2026 12:18:08 -0800 Subject: [PATCH 3/5] claude change --- .../controller/bluegreen/BlueGreenDeploymentService.java | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) 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 06f1915d92..e2c94979c8 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 @@ -750,8 +750,11 @@ private UpdateControl abortDeployment( // Clear stale savepointTriggerId to prevent reuse after abort (defense-in-depth). context.getDeploymentStatus().setSavepointTriggerId(null); - // Autoscaler overrides persist in the active child's lastReconciledSpec - // and will be re-detected on the next ACTIVE reconcile. + // Autoscaler overrides persist in the active child's lastReconciledSpec. + // They will NOT be re-detected immediately because markDeploymentFailing sets + // FAILING, and the FAILING guard in checkAndInitiateDeployment blocks autoscaler + // checks — preventing an infinite abort→re-trigger→abort loop. + // A user spec change is required to clear FAILING and unblock transitions. FlinkBlueGreenDeploymentState previousState = getPreviousState(nextState, context.getDeployments()); @@ -930,7 +933,7 @@ private UpdateControl checkAutoscalerTriggeredTransiti // automatically, including future autoscaler capabilities. var autoscalerDiff = new ReflectiveDiffBuilder<>( - KubernetesDeploymentMode.NATIVE, + KubernetesDeploymentMode.getDeploymentMode(activeChild), childKubeSpec, childLastReconciledSpec) .build(); From 2978c35aa9315d40281134192a4be23086c301a8 Mon Sep 17 00:00:00 2001 From: James Kan Date: Thu, 12 Feb 2026 12:26:58 -0800 Subject: [PATCH 4/5] Cleanup --- .../bluegreen/BlueGreenDeploymentService.java | 80 ++++--------------- .../AbstractFlinkResourceReconciler.java | 35 ++++---- .../utils/bluegreen/BlueGreenUtils.java | 20 +---- 3 files changed, 35 insertions(+), 100 deletions(-) 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 e2c94979c8..560f27473c 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 @@ -75,16 +75,7 @@ public class BlueGreenDeploymentService { // ==================== Deployment Initiation Methods ==================== - /** - * Initiates a new Blue/Green deployment. - * - * @param context the transition context - * @param nextBlueGreenDeploymentType the type of deployment to create - * @param nextState the next state to transition to - * @param lastCheckpoint the checkpoint to restore from (can be null) - * @param isFirstDeployment whether this is the first deployment - * @return UpdateControl for the deployment - */ + /** Convenience overload without spec override. */ public UpdateControl initiateDeployment( BlueGreenContext context, BlueGreenDeploymentType nextBlueGreenDeploymentType, @@ -101,11 +92,8 @@ public UpdateControl initiateDeployment( } /** - * Initiates a new Blue/Green deployment, optionally using a pre-built spec override. - * - * @param specOverride if non-null, used as the source spec for child creation instead of the - * parent's current spec. This is used by autoscaler-triggered transitions to pass overrides - * without mutating the parent's in-memory spec. + * Initiates a B/G deployment. If {@code specOverride} is non-null, the new child is created + * from it instead of the parent's current spec (used for autoscaler-triggered transitions). */ private UpdateControl initiateDeployment( BlueGreenContext context, @@ -908,38 +896,28 @@ private UpdateControl checkAutoscalerTriggeredTransiti BlueGreenContext context, BlueGreenDeploymentType currentType) { FlinkDeployment activeChild = context.getDeploymentByType(currentType); - if (activeChild == null || activeChild.getStatus() == null) { + if (activeChild == null || !isChildStableForAutoscalerPropagation(activeChild)) { return null; } - if (!isChildStableForAutoscalerPropagation(activeChild)) { - return null; - } - - var reconStatus = activeChild.getStatus().getReconciliationStatus(); - if (reconStatus == null || reconStatus.isBeforeFirstDeployment()) { - return null; - } - - FlinkDeploymentSpec childLastReconciledSpec = reconStatus.deserializeLastReconciledSpec(); + // Stability guarantees reconStatus is non-null and not before first deployment. + FlinkDeploymentSpec childLastReconciledSpec = + activeChild.getStatus().getReconciliationStatus().deserializeLastReconciledSpec(); if (childLastReconciledSpec == null) { return null; } - FlinkDeploymentSpec childKubeSpec = activeChild.getSpec(); - - // Use the same diff infrastructure as the child reconciler to detect overrides. - // This keeps both layers in sync — any field the autoscaler modifies is caught - // automatically, including future autoscaler capabilities. + // Reuse the same ReflectiveDiffBuilder as the child reconciler — any field the + // autoscaler touches is detected automatically, including future capabilities. var autoscalerDiff = new ReflectiveDiffBuilder<>( KubernetesDeploymentMode.getDeploymentMode(activeChild), - childKubeSpec, + activeChild.getSpec(), childLastReconciledSpec) .build(); if (autoscalerDiff.getType() == DiffType.IGNORE) { - return null; // No autoscaler changes — nothing to do + return null; } LOG.info( @@ -948,9 +926,8 @@ private UpdateControl checkAutoscalerTriggeredTransiti activeChild.getMetadata().getName(), autoscalerDiff.getType()); - // Handle savepoint if required — reuses the existing savepoint flow. - // Do NOT call setLastReconciledSpec before SAVEPOINTING so overrides - // are re-detected from the child's lastReconciledSpec after completion. + // Savepoint first if required. Don't setLastReconciledSpec yet — overrides + // must remain detectable from lastReconciledSpec after savepoint completes. try { boolean savepointTriggered = handleSavepoint(context, activeChild); if (savepointTriggered) { @@ -965,13 +942,7 @@ private UpdateControl checkAutoscalerTriggeredTransiti return markDeploymentFailing(context, error, e); } - // Serialize the clean parent spec (without overrides) as the lastReconciledSpec. setLastReconciledSpec(context); - - // Build a merged spec: deep-copy the parent and swap in the child's - // lastReconciledSpec as the template. Since we only reach this path when - // there are no user spec changes, the child's lastReconciledSpec already - // equals the parent's template spec + all autoscaler modifications. FlinkBlueGreenDeploymentSpec mergedSpec = buildMergedSpec(context, childLastReconciledSpec); try { @@ -984,38 +955,21 @@ private UpdateControl checkAutoscalerTriggeredTransiti } /** - * Builds a spec for the new child by deep-copying the parent and overwriting only the fields - * the autoscaler modifies ({@code flinkConfiguration} and {@code taskManager}). - * - *

We copy these fields wholesale from the child's {@code lastReconciledSpec} rather - * than computing per-key deltas. Because we only reach this path when there are no user spec - * changes, the child's config/TM equals {@code parentValue + autoscalerDelta}; copying the - * whole field implicitly handles additions, changes, and removals. - * - *

We intentionally do not replace the entire {@code template.spec} because the - * child's {@link FlinkDeploymentSpec} contains deployment-specific transformations applied by - * {@code prepareFlinkDeployment} (e.g., ingress template prefixed with {@code blue-}/{@code - * green-}) that must not leak back into the parent's template. + * Deep-copies the parent spec and overwrites the autoscaler-managed fields ({@code + * flinkConfiguration}, {@code taskManager}) wholesale from the child's {@code + * lastReconciledSpec}. We don't replace the entire {@code template.spec} because it contains + * deployment-specific transforms (e.g., ingress prefixing). */ private FlinkBlueGreenDeploymentSpec buildMergedSpec( BlueGreenContext context, FlinkDeploymentSpec childLastReconciledSpec) { - - // Deep copy the parent spec via JSON round-trip to avoid mutation. FlinkBlueGreenDeploymentSpec merged = SpecUtils.readSpecFromJSON( SpecUtils.writeSpecAsJSON(context.getBgDeployment().getSpec(), "spec"), "spec", FlinkBlueGreenDeploymentSpec.class); - - // Overwrite only the autoscaler-managed fields (KubernetesScalingRealizer touches - // flinkConfiguration and taskManager.resource.memory — nothing else). - // Copying wholesale avoids per-key delta logic and naturally propagates config - // key removals (e.g., from memory tuning) that would otherwise cause phantom - // transition loops. var parentFlinkSpec = merged.getTemplate().getSpec(); parentFlinkSpec.setFlinkConfiguration(childLastReconciledSpec.getFlinkConfiguration()); parentFlinkSpec.setTaskManager(childLastReconciledSpec.getTaskManager()); - return merged; } diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java index 3b08f3f36f..46dd463aeb 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java +++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java @@ -130,21 +130,18 @@ public void reconcile(FlinkResourceContext ctx) throws Exception { cr.getStatus().getReconciliationStatus().deserializeLastReconciledSpec(); SPEC currentDeploySpec = cr.getSpec(); - // For B/G owned children, capture whether there's a user/parent-initiated spec change - // BEFORE the autoscaler modifies the in-memory spec. This lets us distinguish - // autoscaler-induced changes from parent-initiated changes. + // Snapshot pre-autoscaler diff for B/G children so we can distinguish + // autoscaler-only changes from parent-initiated ones after applyAutoscaler(). boolean isBlueGreenOwned = isBlueGreenOwnedDeployment(ctx); - boolean hasPreAutoscalerSpecChange = false; - if (isBlueGreenOwned) { - hasPreAutoscalerSpecChange = - DiffType.IGNORE - != new ReflectiveDiffBuilder<>( - ctx.getDeploymentMode(), - lastReconciledSpec, - currentDeploySpec) - .build() - .getType(); - } + boolean hasPreAutoscalerSpecChange = + isBlueGreenOwned + && DiffType.IGNORE + != new ReflectiveDiffBuilder<>( + ctx.getDeploymentMode(), + lastReconciledSpec, + currentDeploySpec) + .build() + .getType(); applyAutoscaler(ctx); @@ -170,15 +167,11 @@ public void reconcile(FlinkResourceContext ctx) throws Exception { return; } - // For B/G owned children: if the diff is PURELY from the autoscaler - // (no pre-existing spec change from parent), skip ALL in-place processing. - // This covers both horizontal (SCALE) and vertical (UPGRADE) autoscaler changes. - // The B/G parent controller will detect this via lastReconciledSpec and trigger - // a Blue/Green transition instead. + // Autoscaler-only diff on a B/G child → skip in-place processing. + // The B/G parent will detect this via lastReconciledSpec and trigger a transition. if (isBlueGreenOwned && !hasPreAutoscalerSpecChange) { LOG.info( - "Skipping autoscaler-induced spec change ({}) for B/G owned '{}'. " - + "Changes will be handled via Blue/Green transition.", + "Deferring autoscaler change ({}) on B/G child '{}' to parent transition.", diffType, cr.getMetadata().getName()); ReconciliationUtils.updateStatusForDeployedSpec( 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 b17fbc22d0..dd3d837abc 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 @@ -300,17 +300,7 @@ public static Savepoint getLastCheckpoint( // ==================== Deployment Preparation Utilities ==================== - /** - * Creates a new FlinkDeployment resource for a Blue/Green deployment transition. This method - * prepares the deployment with proper metadata, specs, and savepoint configuration. - * - * @param context the Blue/Green transition context - * @param blueGreenDeploymentType the type of deployment (BLUE or GREEN) - * @param lastCheckpoint the savepoint/checkpoint to restore from (can be null) - * @param isFirstDeployment whether this is the initial deployment - * @param bgMeta the metadata of the parent Blue/Green deployment - * @return configured FlinkDeployment ready for deployment - */ + /** Convenience overload without spec override. */ public static FlinkDeployment prepareFlinkDeployment( BlueGreenContext context, BlueGreenDeploymentType blueGreenDeploymentType, @@ -322,11 +312,9 @@ public static FlinkDeployment prepareFlinkDeployment( } /** - * Creates a new FlinkDeployment resource for a Blue/Green deployment transition, optionally - * using a pre-built spec override (e.g., with autoscaler overrides already applied). - * - * @param specOverride if non-null, used as the source spec instead of the parent's current - * spec. This avoids mutating the parent's in-memory spec for ephemeral overrides. + * Creates a FlinkDeployment for a B/G transition. If {@code specOverride} is non-null, it is + * used as the source spec instead of the parent's current spec (e.g., for autoscaler + * overrides). */ public static FlinkDeployment prepareFlinkDeployment( BlueGreenContext context, From 2f432dadd43eb081aeca8095a291cd61085c56d7 Mon Sep 17 00:00:00 2001 From: James Kan Date: Thu, 12 Feb 2026 12:31:25 -0800 Subject: [PATCH 5/5] redudnancy cleanup --- .../AbstractFlinkResourceReconciler.java | 17 ++++------------- 1 file changed, 4 insertions(+), 13 deletions(-) diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java index 46dd463aeb..ae1b56f6a0 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java +++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java @@ -132,7 +132,10 @@ public void reconcile(FlinkResourceContext ctx) throws Exception { // Snapshot pre-autoscaler diff for B/G children so we can distinguish // autoscaler-only changes from parent-initiated ones after applyAutoscaler(). - boolean isBlueGreenOwned = isBlueGreenOwnedDeployment(ctx); + boolean isBlueGreenOwned = + cr instanceof FlinkDeployment + && BlueGreenKubernetesService.isOwnedByBlueGreenDeployment( + (FlinkDeployment) cr); boolean hasPreAutoscalerSpecChange = isBlueGreenOwned && DiffType.IGNORE @@ -396,18 +399,6 @@ private boolean scale(FlinkResourceContext ctx, Configuration deployConfig) return scaled; } - /** - * Checks if the current resource is a FlinkDeployment owned by a FlinkBlueGreenDeployment. - * - * @param ctx the resource context - * @return true if the resource is a B/G owned FlinkDeployment - */ - private boolean isBlueGreenOwnedDeployment(FlinkResourceContext ctx) { - return ctx.getResource() instanceof FlinkDeployment - && BlueGreenKubernetesService.isOwnedByBlueGreenDeployment( - (FlinkDeployment) ctx.getResource()); - } - /** * Checks whether the currently deployed Flink resource spec should be rolled back to the stable * spec. This includes validating the current deployment status, config and checking if the last