[FLINK-35052] Reject unsupported versions in the webhook validator

Author: afedulov

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java

@@ -30,6 +30,7 @@
 import org.apache.flink.kubernetes.operator.api.FlinkSessionJob;
 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;
 import org.apache.flink.kubernetes.operator.api.spec.IngressSpec;
 import org.apache.flink.kubernetes.operator.api.spec.JobManagerSpec;
 import org.apache.flink.kubernetes.operator.api.spec.JobSpec;
@@ -133,10 +134,16 @@ private Optional<String> validateDeploymentName(String name) {
 
     private Optional<String> validateFlinkVersion(FlinkDeployment deployment) {
         var spec = deployment.getSpec();
-        if (spec.getFlinkVersion() == null) {
+        var version = spec.getFlinkVersion();
+        if (version == null) {
             return Optional.of("Flink Version must be defined.");
         }
 
+        if (!FlinkVersion.isSupported(version)) {
+            return Optional.of(
+                    "Flink version " + version + " is not supported by this operator version");
+        }
+
         var lastReconciledSpec =
                 deployment.getStatus().getReconciliationStatus().deserializeLastReconciledSpec();
 

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

@@ -483,6 +483,12 @@ public void testValidationWithoutDefaultConfig() {
 
         testError(dep -> dep.getSpec().setFlinkVersion(null), "Flink Version must be defined.");
 
+        testError(
+                dep -> dep.getSpec().setFlinkVersion(FlinkVersion.v1_14),
+                "Flink version "
+                        + FlinkVersion.v1_14
+                        + " is not supported by this operator version");
+
         testSuccess(dep -> dep.getSpec().setFlinkVersion(FlinkVersion.v1_15));
 
         testError(