[FLINK-38479] Problems after upgrading the operator from 1.12.1 to 1.13.0

Author: csviri

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

@@ -86,8 +86,14 @@ private void appendFields(final Class<?> clazz) {
                     var leftField = readField(field, before, true);
                     var rightField = readField(field, after, true);
                     if (field.getName().equals(FLINK_CONFIGURATION_PROPERTY_NAME)) {
-                        leftField = ((ConfigObjectNode) leftField).asFlatMap();
-                        rightField = ((ConfigObjectNode) rightField).asFlatMap();
+                        leftField =
+                                leftField == null
+                                        ? null
+                                        : ((ConfigObjectNode) leftField).asFlatMap();
+                        rightField =
+                                rightField == null
+                                        ? null
+                                        : ((ConfigObjectNode) rightField).asFlatMap();
                     }
 
                     if (field.isAnnotationPresent(SpecDiff.Config.class)

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

@@ -22,6 +22,7 @@
 import org.apache.flink.configuration.PipelineOptions;
 import org.apache.flink.kubernetes.operator.TestUtils;
 import org.apache.flink.kubernetes.operator.api.diff.DiffType;
+import org.apache.flink.kubernetes.operator.api.spec.ConfigObjectNode;
 import org.apache.flink.kubernetes.operator.api.spec.FlinkDeploymentSpec;
 import org.apache.flink.kubernetes.operator.api.spec.FlinkSessionJobSpec;
 import org.apache.flink.kubernetes.operator.api.spec.FlinkVersion;
@@ -402,6 +403,19 @@ public void testSavepointNonceDiff() {
         assertEquals(0, diff.getNumDiffs());
     }
 
+    @Test
+    public void testNullFlinkConfiguration() {
+        var left = new FlinkDeploymentSpec();
+        left.setFlinkConfiguration((ConfigObjectNode) null);
+        var right = new FlinkDeploymentSpec();
+
+        // This should fail with NullPointerException because there's no null check
+        // in ReflectiveDiffBuilder when casting flinkConfiguration to ConfigObjectNode
+        var diff =
+                new ReflectiveDiffBuilder<>(KubernetesDeploymentMode.NATIVE, left, right).build();
+        assertEquals(DiffType.IGNORE, diff.getType());
+    }
+
     @Value
     private static class TestClass {
         boolean[] f0;