Removing redundant BlueGreenDiffType cases

Author: schongloo

flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/bluegreen/BlueGreenDiffType.java

@@ -26,12 +26,5 @@ public enum BlueGreenDiffType {
     TRANSITION,
 
     /** Changes that only affect the child FlinkDeploymentSpec. */
-    PATCH_CHILD,
-
-    // TODO: the PATCH_TOP_LEVEL and PATCH_BOTH values are redundant, eliminate
-    /** Changes that only affect the top-level configuration. */
-    PATCH_TOP_LEVEL,
-
-    /** Changes that affect both top-level and child specifications. */
-    PATCH_BOTH
+    PATCH_CHILD
 }

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/api/bluegreen/BlueGreenDiffType.java

@@ -146,7 +146,7 @@ private UpdateControl<FlinkBlueGreenDeployment> patchFlinkDeployment(
             DeploymentType currentDeploymentType,
             BlueGreenDiffType specDiff) {
 
-        if (specDiff == BlueGreenDiffType.PATCH_BOTH || specDiff == BlueGreenDiffType.PATCH_CHILD) {
+        if (specDiff == BlueGreenDiffType.PATCH_CHILD) {
             FlinkDeployment nextFlinkDeployment =
                     context.getDeploymentByType(currentDeploymentType);
 

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/bluegreen/BlueGreenDeploymentService.java

@@ -63,7 +63,6 @@ public BlueGreenDiffType compare() {
             return BlueGreenDiffType.IGNORE;
         }
 
-        boolean hasTopLevelDiff = hasTopLevelDifferences();
         BlueGreenDiffType childDiffType = getChildSpecDiffType();
 
         // If nested spec has SCALE or UPGRADE differences, return TRANSITION regardless
@@ -74,13 +73,11 @@ public BlueGreenDiffType compare() {
         // Determine result based on where differences are found
         boolean hasChildDiff = childDiffType != BlueGreenDiffType.IGNORE;
 
-        if (hasTopLevelDiff && hasChildDiff) {
-            return BlueGreenDiffType.PATCH_BOTH;
-        } else if (hasTopLevelDiff) {
-            return BlueGreenDiffType.PATCH_TOP_LEVEL;
-        } else if (hasChildDiff) {
+        if (hasChildDiff) {
+            // Child spec changes take precedence, return the child diff type
             return BlueGreenDiffType.PATCH_CHILD;
-        } else {
+        }
+        else {
             return BlueGreenDiffType.IGNORE;
         }
     }

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/diff/FlinkBlueGreenDeploymentSpecDiff.java

@@ -281,7 +281,8 @@ public void verifyFailureDuringTransition(FlinkVersion flinkVersion) throws Exce
             assertFalse(rs.updateControl.isPatchResource());
             assertTrue(rs.updateControl.getScheduleDelay().isPresent());
             reschedDelayMs = rs.updateControl.getScheduleDelay().get();
-            assertTrue(reschedDelayMs == reconciliationReschedulingIntervalMs && reschedDelayMs > 0);
+            assertTrue(
+                    reschedDelayMs == reconciliationReschedulingIntervalMs && reschedDelayMs > 0);
             assertTrue(
                     instantStrToMillis(rs.reconciledStatus.getAbortTimestamp())
                             > System.currentTimeMillis());
@@ -409,12 +410,18 @@ public void verifyPatchScenario(FlinkVersion flinkVersion, PatchTestCase testCas
 
         testCase.applyChanges(rs.deployment, kubernetesClient);
 
-        var result = reconcileAndVerifyPatchBehavior(rs);
-
-        testCase.verifySpecificBehavior(result, getFlinkDeployments());
-
-        assertFinalized(
-                result.minReconciliationTs, result.rs, FlinkBlueGreenDeploymentState.ACTIVE_BLUE);
+        // PatchTopLevelTestCase should now be ignored (return noUpdate)
+        if (testCase instanceof PatchTopLevelTestCase) {
+            var result = reconcileAndVerifyIgnoreBehavior(rs);
+            testCase.verifySpecificBehavior(result, getFlinkDeployments());
+        } else {
+            var result = reconcileAndVerifyPatchBehavior(rs);
+            testCase.verifySpecificBehavior(result, getFlinkDeployments());
+            assertFinalized(
+                    result.minReconciliationTs,
+                    result.rs,
+                    FlinkBlueGreenDeploymentState.ACTIVE_BLUE);
+        }
     }
 
     // ==================== Parameterized Test Inputs ====================
@@ -607,6 +614,30 @@ private ReconcileResult reconcileAndVerifyPatchBehavior(
         return new ReconcileResult(minReconciliationTs, rs, existingFlinkDeployment);
     }
 
+    private ReconcileResult reconcileAndVerifyIgnoreBehavior(
+            TestingFlinkBlueGreenDeploymentController.BlueGreenReconciliationResult rs)
+            throws Exception {
+
+        var flinkDeployments = getFlinkDeployments();
+        assertEquals(1, flinkDeployments.size());
+        var existingFlinkDeployment = flinkDeployments.get(0);
+
+        var minReconciliationTs = System.currentTimeMillis() - 1;
+        rs = reconcile(rs.deployment);
+
+        assertIgnoreOperationTriggered(rs);
+
+        return new ReconcileResult(minReconciliationTs, rs, existingFlinkDeployment);
+    }
+
+    private void assertIgnoreOperationTriggered(
+            TestingFlinkBlueGreenDeploymentController.BlueGreenReconciliationResult rs) {
+        // For IGNORE behavior, we expect noUpdate (no patch status, no reschedule)
+        assertFalse(rs.updateControl.isPatchStatus());
+        assertFalse(rs.updateControl.isPatchResource());
+        assertFalse(rs.updateControl.getScheduleDelay().isPresent());
+    }
+
     private void assertPatchOperationTriggered(
             TestingFlinkBlueGreenDeploymentController.BlueGreenReconciliationResult rs,
             long minReconciliationTs) {

flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkBlueGreenDeploymentControllerTest.java

@@ -101,7 +101,7 @@ public void testIgnoreForIdenticalSpecs() {
     }
 
     @Test
-    public void testPatchTopLevelForMetadataDifference() {
+    public void testIgnoreForMetadataDifference() {
         FlinkBlueGreenDeploymentSpec spec1 = createBasicSpec();
         FlinkBlueGreenDeploymentSpec spec2 = createBasicSpec();
 
@@ -113,11 +113,11 @@ public void testPatchTopLevelForMetadataDifference() {
         FlinkBlueGreenDeploymentSpecDiff diff =
                 new FlinkBlueGreenDeploymentSpecDiff(DEPLOYMENT_MODE, spec1, spec2);
 
-        assertEquals(BlueGreenDiffType.PATCH_TOP_LEVEL, diff.compare());
+        assertEquals(BlueGreenDiffType.IGNORE, diff.compare());
     }
 
     @Test
-    public void testPatchTopLevelForConfigurationDifference() {
+    public void testIgnoreForConfigurationDifference() {
         FlinkBlueGreenDeploymentSpec spec1 = createBasicSpec();
         FlinkBlueGreenDeploymentSpec spec2 = createBasicSpec();
 
@@ -129,7 +129,7 @@ public void testPatchTopLevelForConfigurationDifference() {
         FlinkBlueGreenDeploymentSpecDiff diff =
                 new FlinkBlueGreenDeploymentSpecDiff(DEPLOYMENT_MODE, spec1, spec2);
 
-        assertEquals(BlueGreenDiffType.PATCH_TOP_LEVEL, diff.compare());
+        assertEquals(BlueGreenDiffType.IGNORE, diff.compare());
     }
 
     @Test
@@ -150,7 +150,7 @@ public void testPatchChildForNestedSpecDifference() {
     }
 
     @Test
-    public void testPatchBothForTopLevelAndNestedDifferences() {
+    public void testPatchChildForTopLevelAndNestedDifferences() {
         FlinkBlueGreenDeploymentSpec spec1 = createBasicSpec();
         FlinkBlueGreenDeploymentSpec spec2 = createBasicSpec();
 
@@ -166,7 +166,7 @@ public void testPatchBothForTopLevelAndNestedDifferences() {
         FlinkBlueGreenDeploymentSpecDiff diff =
                 new FlinkBlueGreenDeploymentSpecDiff(DEPLOYMENT_MODE, spec1, spec2);
 
-        assertEquals(BlueGreenDiffType.PATCH_BOTH, diff.compare());
+        assertEquals(BlueGreenDiffType.PATCH_CHILD, diff.compare());
     }
 
     @Test
@@ -198,7 +198,7 @@ public void testTransitionForUpgradeDifference() {
     }
 
     @Test
-    public void testTransitionOverridesPatchBoth() {
+    public void testTransitionOverridesPatchChild() {
         FlinkBlueGreenDeploymentSpec spec1 = createBasicSpec();
         FlinkBlueGreenDeploymentSpec spec2 = createBasicSpec();
 
@@ -211,7 +211,7 @@ public void testTransitionOverridesPatchBoth() {
         FlinkBlueGreenDeploymentSpecDiff diff =
                 new FlinkBlueGreenDeploymentSpecDiff(DEPLOYMENT_MODE, spec1, spec2);
 
-        // Should return TRANSITION, not PATCH_BOTH
+        // Should return TRANSITION, not PATCH_CHILD
         assertEquals(BlueGreenDiffType.TRANSITION, diff.compare());
     }