This is an automated email from the ASF dual-hosted git repository.
gyfora pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git
The following commit(s) were added to refs/heads/main by this push:
new 567ac1cca [FLINK-40092] Finalize BlueGreen transition after deleting
previous deployment (#1154)
567ac1cca is described below
commit 567ac1cca7987df19308e519715aca0fe71f4005
Author: Farooq Qaiser <[email protected]>
AuthorDate: Tue Jul 14 16:07:59 2026 -0400
[FLINK-40092] Finalize BlueGreen transition after deleting previous
deployment (#1154)
---
.../bluegreen/BlueGreenDeploymentService.java | 11 ++-
.../FlinkBlueGreenDeploymentControllerTest.java | 86 ++++++++++++++++++++++
2 files changed, 93 insertions(+), 4 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 931bf36e5..4e6621545 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
@@ -608,7 +608,7 @@ public class BlueGreenDeploymentService {
long deletionTimestamp = deploymentReadyTimestamp +
deploymentDeletionDelayMs;
if (deletionTimestamp < System.currentTimeMillis()) {
- return deleteDeployment(currentDeployment, context);
+ return deleteDeployment(currentDeployment, context, nextState);
} else {
return waitBeforeDeleting(currentDeployment, deletionTimestamp);
}
@@ -627,17 +627,20 @@ public class BlueGreenDeploymentService {
}
private UpdateControl<FlinkBlueGreenDeployment> deleteDeployment(
- FlinkDeployment currentDeployment, BlueGreenContext context) {
+ FlinkDeployment currentDeployment,
+ BlueGreenContext context,
+ FlinkBlueGreenDeploymentState nextState) {
boolean deleted = deleteFlinkDeployment(currentDeployment, context);
if (!deleted) {
LOG.info("FlinkDeployment '{}' not deleted, will retry",
currentDeployment);
+ return UpdateControl.<FlinkBlueGreenDeployment>noUpdate()
+ .rescheduleAfter(RETRY_DELAY_MS);
} else {
LOG.info("FlinkDeployment '{}' deleted!", currentDeployment);
+ return finalizeBlueGreenDeployment(context, nextState);
}
-
- return
UpdateControl.<FlinkBlueGreenDeployment>noUpdate().rescheduleAfter(RETRY_DELAY_MS);
}
// ==================== Abort and Retry Methods ====================
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 b3b348ded..5a9d76f46 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
@@ -57,11 +57,13 @@ import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
import java.io.IOException;
+import java.time.Duration;
import java.time.Instant;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Stream;
import static
org.apache.flink.kubernetes.operator.api.spec.FlinkBlueGreenDeploymentConfigOptions.ABORT_GRACE_PERIOD;
@@ -71,6 +73,7 @@ import static
org.apache.flink.kubernetes.operator.api.utils.BaseTestUtils.SAMPL
import static
org.apache.flink.kubernetes.operator.api.utils.BaseTestUtils.TEST_DEPLOYMENT_NAME;
import static
org.apache.flink.kubernetes.operator.api.utils.BaseTestUtils.TEST_NAMESPACE;
import static
org.apache.flink.kubernetes.operator.utils.bluegreen.BlueGreenUtils.instantStrToMillis;
+import static org.awaitility.Awaitility.await;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
@@ -558,6 +561,82 @@ public class FlinkBlueGreenDeploymentControllerTest {
testTransitionToGreen(rs, customValue, null);
}
+ @ParameterizedTest
+
@MethodSource("org.apache.flink.kubernetes.operator.TestUtils#flinkVersions")
+ public void verifyGreenNotAbortedWhenNotReadyAfterBlueDeleted(FlinkVersion
flinkVersion)
+ throws Exception {
+ var blueGreenDeployment =
+ buildSessionCluster(
+ TEST_DEPLOYMENT_NAME,
+ TEST_NAMESPACE,
+ flinkVersion,
+ null,
+ UpgradeMode.STATELESS);
+ // Tiny abort grace so the abort deadline is already expired by the
time Blue is deleted.
+ var configuration = blueGreenDeployment.getSpec().getConfiguration();
+ configuration.put(ABORT_GRACE_PERIOD.key(), "1");
+ configuration.put(DEPLOYMENT_DELETION_DELAY.key(), "0");
+
+ blueGreenDeployment
+ .getSpec()
+ .setIngress(
+ IngressSpec.builder()
+ .template("{{name}}.{{namespace}}.example.com")
+ .className("nginx")
+ .build());
+
+ var rs = executeBasicDeployment(flinkVersion, blueGreenDeployment,
false, null);
+ assertIngressPointsToService(BLUE_CLUSTER_ID + REST_SVC_NAME_SUFFIX);
+
+ // Trigger the transition to Green
+ simulateChangeInSpec(rs.deployment, UUID.randomUUID().toString(), 0,
null);
+ rs = reconcile(rs.deployment);
+ assertEquals(
+ FlinkBlueGreenDeploymentState.TRANSITIONING_TO_GREEN,
+ rs.reconciledStatus.getBlueGreenState());
+ assertEquals(2, getFlinkDeployments().size());
+
+ // Green becomes ready -> marks the ready timestamp and schedules
deletion of Blue.
+ simulateSuccessfulJobStart(getFlinkDeploymentByName(GREEN_CLUSTER_ID));
+
+ // Reconcile until the (zero) deletion delay elapses: Blue is deleted
at cutover and the
+ // transition finalizes forward to ACTIVE_GREEN.
+ var latestRs = new AtomicReference<>(rs);
+ await().atMost(Duration.ofSeconds(10))
+ .until(
+ () -> {
+ latestRs.set(reconcile(latestRs.get().deployment));
+ return
latestRs.get().reconciledStatus.getBlueGreenState()
+ ==
FlinkBlueGreenDeploymentState.ACTIVE_GREEN;
+ });
+ rs = latestRs.get();
+
+ assertEquals(1, getFlinkDeployments().size(), "Blue should be deleted
at cutover");
+ assertEquals(
+ FlinkBlueGreenDeploymentState.ACTIVE_GREEN,
+ rs.reconciledStatus.getBlueGreenState(),
+ "Transition must commit forward to ACTIVE_GREEN when Blue is
deleted");
+ assertEquals(
+ 0,
+ instantStrToMillis(rs.reconciledStatus.getAbortTimestamp()),
+ "Abort timer must be cleared once finalized");
+ assertIngressPointsToService(GREEN_CLUSTER_ID + REST_SVC_NAME_SUFFIX);
+
+ // Post-cutover restart: Green briefly goes not-ready
+ simulateJobFailure(getFlinkDeploymentByName(GREEN_CLUSTER_ID));
+ rs = reconcile(rs.deployment);
+
+ assertEquals(1, getFlinkDeployments().size(), "Green must still
exist");
+ assertNotEquals(
+ JobState.SUSPENDED,
+
getFlinkDeploymentByName(GREEN_CLUSTER_ID).getSpec().getJob().getState(),
+ "Green must not be suspended by a post-cutover not-ready
blip");
+ assertEquals(
+ FlinkBlueGreenDeploymentState.ACTIVE_GREEN,
+ rs.reconciledStatus.getBlueGreenState(),
+ "Must remain ACTIVE_GREEN; must not roll back after cutover");
+ }
+
private static String getFlinkConfigurationValue(
FlinkDeploymentSpec flinkDeploymentSpec, String propertyName) {
return
flinkDeploymentSpec.getFlinkConfiguration().get(propertyName).asText();
@@ -1474,6 +1553,13 @@ public class FlinkBlueGreenDeploymentControllerTest {
.getItems();
}
+ private FlinkDeployment getFlinkDeploymentByName(String name) {
+ return getFlinkDeployments().stream()
+ .filter(d -> name.equals(d.getMetadata().getName()))
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("FlinkDeployment '" +
name + "' not found"));
+ }
+
private static FlinkBlueGreenDeployment buildSessionCluster(
String name,
String namespace,
