[FLINK-38517] Fixed equal check to happen after inplace modifications diffResult

Author: drossos

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

@@ -59,15 +59,17 @@ public BlueGreenDiffType compare() {
         FlinkDeploymentSpec leftSpec = left.getTemplate().getSpec();
         FlinkDeploymentSpec rightSpec = right.getTemplate().getSpec();
 
+        // Used in Case 2 & 3: Delegate to ReflectiveDiffBuilder for nested spec comparison
+        // Calculate diffResult before comparison to apply in-place removal of ignored fields
+        DiffResult<FlinkDeploymentSpec> diffResult =
+                new ReflectiveDiffBuilder<>(deploymentMode, leftSpec, rightSpec).build();
+
         // Case 1: FlinkDeploymentSpecs are identical
         if (leftSpec.equals(rightSpec)) {
             return BlueGreenDiffType.IGNORE;
         }
 
-        // Case 2 & 3: Delegate to ReflectiveDiffBuilder for nested spec comparison
-        DiffResult<FlinkDeploymentSpec> diffResult =
-                new ReflectiveDiffBuilder<>(deploymentMode, leftSpec, rightSpec).build();
-
+        // Extract the diff type from ReflectiveDiffBuilder result
         DiffType diffType = diffResult.getType();
 
         // Case 2: ReflectiveDiffBuilder returns IGNORE

flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/diff/FlinkBlueGreenDeploymentSpecDiffTest.java

@@ -33,6 +33,7 @@
 import org.apache.flink.kubernetes.operator.api.spec.UpgradeMode;
 
 import io.fabric8.kubernetes.api.model.ObjectMeta;
+import io.fabric8.kubernetes.api.model.PodTemplateSpecBuilder;
 import org.junit.jupiter.api.Test;
 
 import java.util.HashMap;
@@ -133,6 +134,68 @@ public void testIgnoreForConfigurationDifference() {
         assertEquals(BlueGreenDiffType.IGNORE, diff.compare());
     }
 
+    @Test
+    public void testIgnoreForRootPodTemplateAdditionalProps() {
+        FlinkBlueGreenDeploymentSpec spec1 = createBasicSpec();
+        FlinkBlueGreenDeploymentSpec spec2 = createBasicSpec();
+
+        // Set spec1 to have empty podTemplate to not cause valid diff to be detected
+        spec1.getTemplate().getSpec().setPodTemplate(new PodTemplateSpecBuilder().build());
+        spec2.getTemplate()
+                .getSpec()
+                .setPodTemplate(
+                        new PodTemplateSpecBuilder()
+                                .withAdditionalProperties(Map.of("apiVersion", "v1"))
+                                .build());
+        FlinkBlueGreenDeploymentSpecDiff diff =
+                new FlinkBlueGreenDeploymentSpecDiff(DEPLOYMENT_MODE, spec1, spec2);
+        assertEquals(BlueGreenDiffType.IGNORE, diff.compare());
+    }
+
+    @Test
+    public void testIgnoreForJobManagerPodTemplateAdditionalProps() {
+        FlinkBlueGreenDeploymentSpec spec1 = createBasicSpec();
+        FlinkBlueGreenDeploymentSpec spec2 = createBasicSpec();
+
+        // Set spec1 to have empty podTemplate to not cause valid diff to be detected
+        spec1.getTemplate()
+                .getSpec()
+                .getJobManager()
+                .setPodTemplate(new PodTemplateSpecBuilder().build());
+        spec2.getTemplate()
+                .getSpec()
+                .getJobManager()
+                .setPodTemplate(
+                        new PodTemplateSpecBuilder()
+                                .withAdditionalProperties(Map.of("apiVersion", "v1"))
+                                .build());
+        FlinkBlueGreenDeploymentSpecDiff diff =
+                new FlinkBlueGreenDeploymentSpecDiff(DEPLOYMENT_MODE, spec1, spec2);
+        assertEquals(BlueGreenDiffType.IGNORE, diff.compare());
+    }
+
+    @Test
+    public void testIgnoreForTaskManagerPodTemplateAdditionalProps() {
+        FlinkBlueGreenDeploymentSpec spec1 = createBasicSpec();
+        FlinkBlueGreenDeploymentSpec spec2 = createBasicSpec();
+
+        // Set spec1 to have empty podTemplate to not cause valid diff to be detected
+        spec1.getTemplate()
+                .getSpec()
+                .getTaskManager()
+                .setPodTemplate(new PodTemplateSpecBuilder().build());
+        spec2.getTemplate()
+                .getSpec()
+                .getTaskManager()
+                .setPodTemplate(
+                        new PodTemplateSpecBuilder()
+                                .withAdditionalProperties(Map.of("apiVersion", "v1"))
+                                .build());
+        FlinkBlueGreenDeploymentSpecDiff diff =
+                new FlinkBlueGreenDeploymentSpecDiff(DEPLOYMENT_MODE, spec1, spec2);
+        assertEquals(BlueGreenDiffType.IGNORE, diff.compare());
+    }
+
     @Test
     public void testTransitionForNestedSpecDifference() {
         FlinkBlueGreenDeploymentSpec spec1 = createBasicSpec();