[FLINK-36577] Add compatibility check for FlinkStateSnapshot (#991)

Author: mateczagany

flink-kubernetes-operator-api/pom.xml

@@ -257,6 +257,23 @@ under the License.
                             </target>
                         </configuration>
                     </execution>
+                    <execution>
+                        <id>statesnapshot-crd-compatibility-check</id>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>run</goal>
+                        </goals>
+                        <configuration>
+                            <target>
+                                <java classname="org.apache.flink.kubernetes.operator.api.validation.CrdCompatibilityChecker"
+                                      fork="true" failonerror="true">
+                                    <classpath refid="maven.compile.classpath"/>
+                                    <arg value="file://${rootDir}/helm/flink-kubernetes-operator/crds/flinkstatesnapshots.flink.apache.org-v1.yml"/>
+                                    <arg value="https://raw.githubusercontent.com/apache/flink-kubernetes-operator/release-1.10.0/helm/flink-kubernetes-operator/crds/flinkstatesnapshots.flink.apache.org-v1.yml"/>
+                                </java>
+                            </target>
+                        </configuration>
+                    </execution>
                 </executions>
             </plugin>
             <plugin>

flink-kubernetes-operator-api/src/test/java/org/apache/flink/kubernetes/operator/api/validation/CrdCompatibilityCheckerTest.java

@@ -19,6 +19,7 @@
 
 import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
 import org.apache.flink.kubernetes.operator.api.FlinkSessionJob;
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -271,6 +272,19 @@ public void testCreateFlinkDeploymentIgnoreUnknownFields() throws IOException {
         assertEquals(flinkDeploymentWithUnknownFields.toString(), flinkDeployment.toString());
     }
 
+    @Test
+    public void testCreateFlinkStateSnapshotIgnoreUnknownFields() throws IOException {
+        FlinkStateSnapshot flinkSnapshotWithUnknownFields =
+                objectMapper.readValue(
+                        new File("src/test/resources/test-snapshot-with-unknown-fields.yaml"),
+                        FlinkStateSnapshot.class);
+        FlinkStateSnapshot flinkSnapshot =
+                objectMapper.readValue(
+                        new File("src/test/resources/test-snapshot.yaml"),
+                        FlinkStateSnapshot.class);
+        assertEquals(flinkSnapshotWithUnknownFields.toString(), flinkSnapshot.toString());
+    }
+
     @Test
     public void testGetSchemaFromUrl() throws IOException {
         var fileUrl =

flink-kubernetes-operator-api/src/test/resources/test-snapshot-with-unknown-fields.yaml

@@ -0,0 +1,30 @@
+################################################################################
+#  Licensed to the Apache Software Foundation (ASF) under one
+#  or more contributor license agreements.  See the NOTICE file
+#  distributed with this work for additional information
+#  regarding copyright ownership.  The ASF licenses this file
+#  to you under the Apache License, Version 2.0 (the
+#  "License"); you may not use this file except in compliance
+#  with the License.  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+apiVersion: flink.apache.org/v1beta1
+kind: FlinkStateSnapshot
+metadata:
+  name: example-savepoint
+spec:
+  backoffLimit: 0
+  jobReference:
+    kind: FlinkDeployment
+    name: example-deployment
+    unknownField: testUnknownField
+  savepoint:
+    unknownField: testUnknownField
+  unknownField: testUnknownField

flink-kubernetes-operator-api/src/test/resources/test-snapshot.yaml

@@ -0,0 +1,27 @@
+################################################################################
+#  Licensed to the Apache Software Foundation (ASF) under one
+#  or more contributor license agreements.  See the NOTICE file
+#  distributed with this work for additional information
+#  regarding copyright ownership.  The ASF licenses this file
+#  to you under the Apache License, Version 2.0 (the
+#  "License"); you may not use this file except in compliance
+#  with the License.  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+apiVersion: flink.apache.org/v1beta1
+kind: FlinkStateSnapshot
+metadata:
+  name: example-savepoint
+spec:
+  backoffLimit: 0
+  jobReference:
+    kind: FlinkDeployment
+    name: example-deployment
+  savepoint: {}