From 3b139b21c22e50737814e26bc4986ba38a1d3f70 Mon Sep 17 00:00:00 2001 From: James Kan Date: Fri, 10 Jul 2026 17:54:13 -0700 Subject: [PATCH 1/2] ObserveGeneration and lastStableGeneration for FlinkBlueGreenDeployment --- .../content/docs/custom-resource/reference.md | 2 + .../FlinkBlueGreenDeploymentStatus.java | 6 + .../FlinkBlueGreenDeploymentController.java | 27 ++++ .../bluegreen/BlueGreenDeploymentService.java | 19 +++ ...linkBlueGreenDeploymentControllerTest.java | 126 ++++++++++++++++++ ...uegreendeployments.flink.apache.org-v1.yml | 4 + 6 files changed, 184 insertions(+) diff --git a/docs/content/docs/custom-resource/reference.md b/docs/content/docs/custom-resource/reference.md index 98d18e7d65..f67743cac8 100644 --- a/docs/content/docs/custom-resource/reference.md +++ b/docs/content/docs/custom-resource/reference.md @@ -361,6 +361,8 @@ This serves as a full reference for FlinkDeployment and FlinkSessionJob custom r | ----------| ---- | ---- | | jobStatus | org.apache.flink.kubernetes.operator.api.status.JobStatus | | | blueGreenState | org.apache.flink.kubernetes.operator.api.status.FlinkBlueGreenDeploymentState | The state of the blue/green transition. | +| observedGeneration | java.lang.Long | Last observed generation of the FlinkBlueGreenDeployment. | +| lastStableGeneration | java.lang.Long | Last generation that reached a stable terminal state. | | lastReconciledSpec | java.lang.String | Last reconciled (serialized) deployment spec. | | lastReconciledTimestamp | java.lang.String | Timestamp of last reconciliation. | | abortTimestamp | java.lang.String | Computed from abortGracePeriodMs, timestamp after which the deployment should be aborted. | diff --git a/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/status/FlinkBlueGreenDeploymentStatus.java b/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/status/FlinkBlueGreenDeploymentStatus.java index 104f195efa..4c86368f5b 100644 --- a/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/status/FlinkBlueGreenDeploymentStatus.java +++ b/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/status/FlinkBlueGreenDeploymentStatus.java @@ -41,6 +41,12 @@ public class FlinkBlueGreenDeploymentStatus { /** The state of the blue/green transition. */ private FlinkBlueGreenDeploymentState blueGreenState; + /** Last observed generation of the FlinkBlueGreenDeployment. */ + private Long observedGeneration; + + /** Last generation that reached a stable terminal state. */ + private Long lastStableGeneration; + /** Last reconciled (serialized) deployment spec. */ private String lastReconciledSpec; diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkBlueGreenDeploymentController.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkBlueGreenDeploymentController.java index a22574470d..b30c4790b0 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkBlueGreenDeploymentController.java +++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkBlueGreenDeploymentController.java @@ -17,6 +17,7 @@ package org.apache.flink.kubernetes.operator.controller; +import org.apache.flink.api.common.JobStatus; import org.apache.flink.kubernetes.operator.api.FlinkBlueGreenDeployment; import org.apache.flink.kubernetes.operator.api.FlinkDeployment; import org.apache.flink.kubernetes.operator.api.status.FlinkBlueGreenDeploymentState; @@ -45,6 +46,8 @@ import java.util.ArrayList; import java.util.List; +import static org.apache.flink.kubernetes.operator.api.status.FlinkBlueGreenDeploymentState.ACTIVE_BLUE; +import static org.apache.flink.kubernetes.operator.api.status.FlinkBlueGreenDeploymentState.ACTIVE_GREEN; import static org.apache.flink.kubernetes.operator.api.status.FlinkBlueGreenDeploymentState.INITIALIZING_BLUE; /** @@ -151,6 +154,30 @@ public UpdateControl reconcile( BlueGreenStateHandler handler = handlerRegistry.getHandler(currentState); UpdateControl updateControl = handler.handle(context); + + var isActiveState = currentState == ACTIVE_BLUE || currentState == ACTIVE_GREEN; + var jobStatus = deploymentStatus.getJobStatus(); + var jobState = jobStatus == null ? null : jobStatus.getState(); + var isTerminalJobState = + jobState == JobStatus.RUNNING + || jobState == JobStatus.FINISHED + || jobState == JobStatus.SUSPENDED; + if (updateControl.isNoUpdate() + && (!BlueGreenDeploymentService.isGenerationObserved(context) + || (isActiveState + && isTerminalJobState + && !BlueGreenDeploymentService.isGenerationStable(context)))) { + if (isActiveState && isTerminalJobState) { + deploymentStatus.setLastStableGeneration( + bgDeployment.getMetadata().getGeneration()); + } + var statusUpdateControl = + BlueGreenDeploymentService.patchStatusUpdateControl( + context, null, null, null); + updateControl.getScheduleDelay().ifPresent(statusUpdateControl::rescheduleAfter); + updateControl = statusUpdateControl; + } + statusRecorder.patchAndCacheStatus(bgDeployment, josdkContext.getClient()); return updateControl; } 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 81bbe0436c..1dca42e078 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 @@ -859,6 +859,18 @@ private static void resetTransitionMarkers(FlinkBlueGreenDeploymentStatus status status.setSavepointTriggerId(null); } + public static boolean isGenerationObserved(BlueGreenContext context) { + return Objects.equals( + context.getDeploymentStatus().getObservedGeneration(), + context.getBgDeployment().getMetadata().getGeneration()); + } + + public static boolean isGenerationStable(BlueGreenContext context) { + return Objects.equals( + context.getDeploymentStatus().getLastStableGeneration(), + context.getBgDeployment().getMetadata().getGeneration()); + } + public static UpdateControl patchStatusUpdateControl( BlueGreenContext context, FlinkBlueGreenDeploymentState deploymentState, @@ -884,6 +896,13 @@ public static UpdateControl patchStatusUpdateControl( deploymentStatus.setError(null); } + if (jobState == JobStatus.RUNNING || jobState == JobStatus.SUSPENDED) { + deploymentStatus.setLastStableGeneration( + flinkBlueGreenDeployment.getMetadata().getGeneration()); + } + + deploymentStatus.setObservedGeneration( + flinkBlueGreenDeployment.getMetadata().getGeneration()); deploymentStatus.setLastReconciledTimestamp(java.time.Instant.now().toString()); flinkBlueGreenDeployment.setStatus(deploymentStatus); return UpdateControl.patchStatus(flinkBlueGreenDeployment); 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 a1a64aa488..5db9d31aa9 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 @@ -52,6 +52,7 @@ import io.javaoperatorsdk.operator.api.reconciler.UpdateControl; import org.jetbrains.annotations.NotNull; import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.Arguments; import org.junit.jupiter.params.provider.MethodSource; @@ -128,6 +129,118 @@ public void verifyBasicDeployment(FlinkVersion flinkVersion, String initialSavep executeBasicDeployment(flinkVersion, blueGreenDeployment, true, initialSavepointPath); } + @Test + public void verifyObservedGenerationUpdatedWhenNewSpecObserved() throws Exception { + var blueGreenDeployment = + buildSessionCluster( + TEST_DEPLOYMENT_NAME, + TEST_NAMESPACE, + FlinkVersion.v1_20, + null, + UpgradeMode.STATELESS); + var rs = executeBasicDeployment(FlinkVersion.v1_20, blueGreenDeployment, false, null); + + Long previousGeneration = rs.deployment.getMetadata().getGeneration(); + assertNotNull(previousGeneration); + assertEquals(previousGeneration, rs.reconciledStatus.getObservedGeneration()); + assertEquals(previousGeneration, rs.reconciledStatus.getLastStableGeneration()); + + simulateSpecChange(rs.deployment, UUID.randomUUID().toString()); + Long newGeneration = previousGeneration + 1; + rs.deployment.getMetadata().setGeneration(newGeneration); + kubernetesClient.resource(rs.deployment).createOrReplace(); + + rs = reconcile(rs.deployment); + + assertEquals(newGeneration, rs.reconciledStatus.getObservedGeneration()); + assertEquals(previousGeneration, rs.reconciledStatus.getLastStableGeneration()); + assertEquals( + FlinkBlueGreenDeploymentState.TRANSITIONING_TO_GREEN, + rs.reconciledStatus.getBlueGreenState()); + assertEquals(JobStatus.RECONCILING, rs.reconciledStatus.getJobStatus().getState()); + } + + @Test + public void verifyObservedGenerationBackfilledForExistingStatus() throws Exception { + var blueGreenDeployment = + buildSessionCluster( + TEST_DEPLOYMENT_NAME, + TEST_NAMESPACE, + FlinkVersion.v1_20, + null, + UpgradeMode.STATELESS); + var rs = executeBasicDeployment(FlinkVersion.v1_20, blueGreenDeployment, false, null); + + rs.deployment.getStatus().setObservedGeneration(null); + rs.deployment.getStatus().setLastStableGeneration(null); + + rs = reconcile(rs.deployment); + + assertTrue(rs.updateControl.isPatchStatus()); + assertEquals( + rs.deployment.getMetadata().getGeneration(), + rs.reconciledStatus.getObservedGeneration()); + assertEquals( + rs.deployment.getMetadata().getGeneration(), + rs.reconciledStatus.getLastStableGeneration()); + assertEquals( + FlinkBlueGreenDeploymentState.ACTIVE_BLUE, rs.reconciledStatus.getBlueGreenState()); + assertEquals(JobStatus.RUNNING, rs.reconciledStatus.getJobStatus().getState()); + } + + @Test + public void verifyStableGenerationNotUpdatedWhenTransitionFailsBeforeStart() throws Exception { + var blueGreenDeployment = + buildSessionCluster( + TEST_DEPLOYMENT_NAME, + TEST_NAMESPACE, + FlinkVersion.v1_20, + null, + UpgradeMode.STATELESS); + var rs = executeBasicDeployment(FlinkVersion.v1_20, blueGreenDeployment, false, null); + + Long previousGeneration = rs.deployment.getMetadata().getGeneration(); + simulateSpecChange(rs.deployment, UUID.randomUUID().toString()); + Long newGeneration = previousGeneration + 1; + rs.deployment.getMetadata().setGeneration(newGeneration); + kubernetesClient.resource(rs.deployment).createOrReplace(); + simulateJobFailure(getFlinkDeployments().get(0)); + + rs = reconcile(rs.deployment); + + assertEquals(newGeneration, rs.reconciledStatus.getObservedGeneration()); + assertEquals(previousGeneration, rs.reconciledStatus.getLastStableGeneration()); + assertEquals(JobStatus.FAILING, rs.reconciledStatus.getJobStatus().getState()); + assertEquals( + FlinkBlueGreenDeploymentState.ACTIVE_BLUE, rs.reconciledStatus.getBlueGreenState()); + } + + @Test + public void verifyStableGenerationNotUpdatedDuringSavepointHandoff() throws Exception { + var blueGreenDeployment = + buildSessionCluster( + TEST_DEPLOYMENT_NAME, + TEST_NAMESPACE, + FlinkVersion.v1_20, + null, + UpgradeMode.SAVEPOINT); + var rs = executeBasicDeployment(FlinkVersion.v1_20, blueGreenDeployment, false, null); + + Long previousGeneration = rs.deployment.getMetadata().getGeneration(); + simulateSpecChange(rs.deployment, UUID.randomUUID().toString()); + Long newGeneration = previousGeneration + 1; + rs.deployment.getMetadata().setGeneration(newGeneration); + kubernetesClient.resource(rs.deployment).createOrReplace(); + + rs = handleSavepoint(rs); + + assertEquals(newGeneration, rs.reconciledStatus.getObservedGeneration()); + assertEquals(previousGeneration, rs.reconciledStatus.getLastStableGeneration()); + assertEquals( + FlinkBlueGreenDeploymentState.ACTIVE_BLUE, rs.reconciledStatus.getBlueGreenState()); + assertEquals(JobStatus.RUNNING, rs.reconciledStatus.getJobStatus().getState()); + } + @ParameterizedTest @MethodSource("flinkVersionsAndSavepointPathsAndUpgradeModes") public void verifyBasicTransition( @@ -1181,6 +1294,12 @@ public void verifyInitialSuspendedIsBlockedThenDeploysOnRunning(FlinkVersion fli JobStatus.SUSPENDED, rs.reconciledStatus.getJobStatus().getState(), "Job status should be SUSPENDED when initial deployment is blocked"); + assertEquals( + rs.deployment.getMetadata().getGeneration(), + rs.reconciledStatus.getObservedGeneration()); + assertEquals( + rs.deployment.getMetadata().getGeneration(), + rs.reconciledStatus.getLastStableGeneration()); // Flip to RUNNING and reconcile again bg = rs.deployment; @@ -1507,6 +1626,12 @@ private void assertFinalized( rs.reconciledStatus.getLastReconciledSpec()); assertEquals(expectedBGDeploymentState, rs.reconciledStatus.getBlueGreenState()); assertEquals(JobStatus.RUNNING, rs.reconciledStatus.getJobStatus().getState()); + assertEquals( + rs.deployment.getMetadata().getGeneration(), + rs.reconciledStatus.getObservedGeneration()); + assertEquals( + rs.deployment.getMetadata().getGeneration(), + rs.reconciledStatus.getLastStableGeneration()); assertEquals(0, instantStrToMillis(rs.reconciledStatus.getDeploymentReadyTimestamp())); assertEquals(0, instantStrToMillis(rs.reconciledStatus.getAbortTimestamp())); @@ -1796,6 +1921,7 @@ private static FlinkBlueGreenDeployment buildSessionCluster( .withCreationTimestamp(Instant.now().toString()) .withUid(UUID.randomUUID().toString()) .withResourceVersion("1") + .withGeneration(1L) .build()); var bgDeploymentSpec = getTestFlinkDeploymentSpec(version); diff --git a/helm/flink-kubernetes-operator/crds/flinkbluegreendeployments.flink.apache.org-v1.yml b/helm/flink-kubernetes-operator/crds/flinkbluegreendeployments.flink.apache.org-v1.yml index 0f8ef636fb..1ea315a672 100644 --- a/helm/flink-kubernetes-operator/crds/flinkbluegreendeployments.flink.apache.org-v1.yml +++ b/helm/flink-kubernetes-operator/crds/flinkbluegreendeployments.flink.apache.org-v1.yml @@ -10938,6 +10938,10 @@ spec: type: "string" lastReconciledTimestamp: type: "string" + lastStableGeneration: + type: "integer" + observedGeneration: + type: "integer" savepointTriggerId: type: "string" type: "object" From a418cc9c1114f254b8b39f1758e64244fd5927a3 Mon Sep 17 00:00:00 2001 From: James Kan Date: Wed, 15 Jul 2026 19:27:18 -0700 Subject: [PATCH 2/2] Code cleanup --- .../FlinkBlueGreenDeploymentController.java | 64 +++++++++++++------ ...linkBlueGreenDeploymentControllerTest.java | 2 +- 2 files changed, 46 insertions(+), 20 deletions(-) diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkBlueGreenDeploymentController.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkBlueGreenDeploymentController.java index b30c4790b0..1c0c18b20d 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkBlueGreenDeploymentController.java +++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkBlueGreenDeploymentController.java @@ -154,33 +154,59 @@ public UpdateControl reconcile( BlueGreenStateHandler handler = handlerRegistry.getHandler(currentState); UpdateControl updateControl = handler.handle(context); + updateControl = + syncGenerationStatusIfNeeded( + bgDeployment, context, currentState, deploymentStatus, updateControl); - var isActiveState = currentState == ACTIVE_BLUE || currentState == ACTIVE_GREEN; - var jobStatus = deploymentStatus.getJobStatus(); - var jobState = jobStatus == null ? null : jobStatus.getState(); - var isTerminalJobState = - jobState == JobStatus.RUNNING - || jobState == JobStatus.FINISHED - || jobState == JobStatus.SUSPENDED; - if (updateControl.isNoUpdate() - && (!BlueGreenDeploymentService.isGenerationObserved(context) - || (isActiveState - && isTerminalJobState - && !BlueGreenDeploymentService.isGenerationStable(context)))) { - if (isActiveState && isTerminalJobState) { - deploymentStatus.setLastStableGeneration( - bgDeployment.getMetadata().getGeneration()); - } + statusRecorder.patchAndCacheStatus(bgDeployment, josdkContext.getClient()); + return updateControl; + } + } + + private static UpdateControl syncGenerationStatusIfNeeded( + FlinkBlueGreenDeployment bgDeployment, + BlueGreenContext context, + FlinkBlueGreenDeploymentState currentState, + FlinkBlueGreenDeploymentStatus deploymentStatus, + UpdateControl updateControl) { + + // Migration/no-op path: kapp may reconcile an existing stable CR without rollout work when + // the rendered BlueGreen spec is unchanged, or when only non-spec resources/metadata + // changed. Handlers return noUpdate() in those cases, so sync generation status here. + if (updateControl.isNoUpdate()) { + var needsSync = !BlueGreenDeploymentService.isGenerationObserved(context); + + if (!BlueGreenDeploymentService.isGenerationStable(context) + && isStableBlueGreenJobStatus(currentState, deploymentStatus)) { + deploymentStatus.setLastStableGeneration( + bgDeployment.getMetadata().getGeneration()); + needsSync = true; + } + + if (needsSync) { var statusUpdateControl = BlueGreenDeploymentService.patchStatusUpdateControl( context, null, null, null); updateControl.getScheduleDelay().ifPresent(statusUpdateControl::rescheduleAfter); - updateControl = statusUpdateControl; + return statusUpdateControl; } + } - statusRecorder.patchAndCacheStatus(bgDeployment, josdkContext.getClient()); - return updateControl; + return updateControl; + } + + private static boolean isStableBlueGreenJobStatus( + FlinkBlueGreenDeploymentState currentState, + FlinkBlueGreenDeploymentStatus deploymentStatus) { + // Suspension is represented as ACTIVE_* with parent jobStatus SUSPENDED, not as a separate + // BlueGreen state. Transitional or failing states must continue waiting for finalization. + if (currentState != ACTIVE_BLUE && currentState != ACTIVE_GREEN) { + return false; } + + var jobStatus = deploymentStatus.getJobStatus(); + var jobState = jobStatus == null ? null : jobStatus.getState(); + return jobState == JobStatus.RUNNING || jobState == JobStatus.SUSPENDED; } public static void logAndThrow(String message) { 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 5db9d31aa9..70bc34282c 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 @@ -161,7 +161,7 @@ public void verifyObservedGenerationUpdatedWhenNewSpecObserved() throws Exceptio } @Test - public void verifyObservedGenerationBackfilledForExistingStatus() throws Exception { + public void verifyGenerationStatusSyncedForExistingStableStatus() throws Exception { var blueGreenDeployment = buildSessionCluster( TEST_DEPLOYMENT_NAME,