From 7a6d3b1268f8ec2160fdd50cc989811249f91878 Mon Sep 17 00:00:00 2001 From: Santwana Verma Date: Mon, 22 Jun 2026 21:40:33 +0530 Subject: [PATCH] [FLINK-31849] Adds basic CRD validations --- .../api/spec/FlinkDeploymentSpec.java | 3 +- .../api/spec/FlinkStateSnapshotSpec.java | 2 + .../operator/api/spec/IngressSpec.java | 3 +- .../operator/api/spec/JobManagerSpec.java | 2 + .../operator/api/spec/TaskManagerSpec.java | 2 + .../FlinkConfigurationYamlSupportTest.java | 67 +++++++++++++++++++ .../operator/validation/DefaultValidator.java | 33 ++++----- .../validation/DefaultValidatorTest.java | 24 +++---- ...uegreendeployments.flink.apache.org-v1.yml | 8 +++ .../flinkdeployments.flink.apache.org-v1.yml | 6 ++ ...linkstatesnapshots.flink.apache.org-v1.yml | 1 + 11 files changed, 114 insertions(+), 37 deletions(-) diff --git a/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/FlinkDeploymentSpec.java b/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/FlinkDeploymentSpec.java index 29e40906ef..aa53827f3a 100644 --- a/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/FlinkDeploymentSpec.java +++ b/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/FlinkDeploymentSpec.java @@ -21,6 +21,7 @@ import com.fasterxml.jackson.annotation.JsonIgnoreProperties; import io.fabric8.crd.generator.annotation.SchemaFrom; +import io.fabric8.generator.annotation.Required; import io.fabric8.kubernetes.api.model.Pod; import io.fabric8.kubernetes.api.model.PodTemplateSpec; import lombok.AllArgsConstructor; @@ -52,7 +53,7 @@ public class FlinkDeploymentSpec extends AbstractFlinkSpec { private String serviceAccount; /** Flink image version. */ - private FlinkVersion flinkVersion; + @Required private FlinkVersion flinkVersion; /** Ingress specs. */ private IngressSpec ingress; diff --git a/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/FlinkStateSnapshotSpec.java b/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/FlinkStateSnapshotSpec.java index 9248d857d6..835b119e48 100644 --- a/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/FlinkStateSnapshotSpec.java +++ b/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/FlinkStateSnapshotSpec.java @@ -20,6 +20,7 @@ import org.apache.flink.annotation.Experimental; import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import io.fabric8.generator.annotation.Min; import lombok.AllArgsConstructor; import lombok.Data; import lombok.NoArgsConstructor; @@ -46,6 +47,7 @@ public class FlinkStateSnapshotSpec { * Maximum number of retries before the snapshot is considered as failed. Set to -1 for * unlimited or 0 for no retries. */ + @Min(-1) private int backoffLimit = -1; public boolean isSavepoint() { diff --git a/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/IngressSpec.java b/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/IngressSpec.java index d5dce55d2a..94ba012302 100644 --- a/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/IngressSpec.java +++ b/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/IngressSpec.java @@ -20,6 +20,7 @@ import org.apache.flink.annotation.Experimental; import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import io.fabric8.generator.annotation.Required; import io.fabric8.kubernetes.api.model.networking.v1.IngressTLS; import lombok.AllArgsConstructor; import lombok.Builder; @@ -39,7 +40,7 @@ public class IngressSpec { /** Ingress template for the JobManager service. */ - private String template; + @Required private String template; /** Ingress className for the Flink deployment. */ private String className; diff --git a/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/JobManagerSpec.java b/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/JobManagerSpec.java index 79943c9c8c..71275de056 100644 --- a/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/JobManagerSpec.java +++ b/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/JobManagerSpec.java @@ -21,6 +21,7 @@ import com.fasterxml.jackson.annotation.JsonIgnoreProperties; import io.fabric8.crd.generator.annotation.SchemaFrom; +import io.fabric8.generator.annotation.Min; import io.fabric8.kubernetes.api.model.Pod; import io.fabric8.kubernetes.api.model.PodTemplateSpec; import io.fabric8.kubernetes.api.model.ResourceRequirements; @@ -51,6 +52,7 @@ public class JobManagerSpec { private ResourceRequirements resources; /** Number of JobManager replicas. Must be 1 for non-HA deployments. */ + @Min(1) private int replicas = 1; /** JobManager pod template. It will be merged with FlinkDeploymentSpec.podTemplate. */ diff --git a/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/TaskManagerSpec.java b/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/TaskManagerSpec.java index 2ee7669e26..ea8c04676d 100644 --- a/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/TaskManagerSpec.java +++ b/flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/TaskManagerSpec.java @@ -24,6 +24,7 @@ import com.fasterxml.jackson.annotation.JsonIgnoreProperties; import io.fabric8.crd.generator.annotation.SchemaFrom; +import io.fabric8.generator.annotation.Min; import io.fabric8.kubernetes.api.model.Pod; import io.fabric8.kubernetes.api.model.PodTemplateSpec; import io.fabric8.kubernetes.api.model.ResourceRequirements; @@ -57,6 +58,7 @@ public class TaskManagerSpec implements Diffable { /** Number of TaskManager replicas. If defined, takes precedence over parallelism */ @SpecDiff(value = DiffType.SCALE, mode = KubernetesDeploymentMode.STANDALONE) @SpecReplicas + @Min(1) private Integer replicas; /** TaskManager pod template. It will be merged with FlinkDeploymentSpec.podTemplate. */ diff --git a/flink-kubernetes-operator-api/src/test/java/org/apache/flink/kubernetes/operator/api/FlinkConfigurationYamlSupportTest.java b/flink-kubernetes-operator-api/src/test/java/org/apache/flink/kubernetes/operator/api/FlinkConfigurationYamlSupportTest.java index 2302938d59..9d75515e85 100644 --- a/flink-kubernetes-operator-api/src/test/java/org/apache/flink/kubernetes/operator/api/FlinkConfigurationYamlSupportTest.java +++ b/flink-kubernetes-operator-api/src/test/java/org/apache/flink/kubernetes/operator/api/FlinkConfigurationYamlSupportTest.java @@ -17,12 +17,17 @@ package org.apache.flink.kubernetes.operator.api; +import org.apache.flink.kubernetes.operator.api.spec.IngressSpec; +import org.apache.flink.kubernetes.operator.api.utils.BaseTestUtils; + import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.dataformat.yaml.YAMLFactory; import io.fabric8.kubeapitest.junit.EnableKubeAPIServer; import io.fabric8.kubernetes.api.model.GenericKubernetesResource; +import io.fabric8.kubernetes.api.model.HasMetadata; import io.fabric8.kubernetes.api.model.apiextensions.v1.CustomResourceDefinition; import io.fabric8.kubernetes.client.KubernetesClient; +import io.fabric8.kubernetes.client.KubernetesClientException; import io.fabric8.kubernetes.client.dsl.NonDeletingOperation; import org.junit.jupiter.api.Test; @@ -30,6 +35,7 @@ import java.io.IOException; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; @EnableKubeAPIServer class FlinkConfigurationYamlSupportTest { @@ -39,6 +45,8 @@ class FlinkConfigurationYamlSupportTest { "src/test/resources/pre-flink-configuration-yaml-crd.yml"; public static final String NEW_CRD_PATH = "../helm/flink-kubernetes-operator/crds/flinkdeployments.flink.apache.org-v1.yml"; + public static final String SNAPSHOT_CRD_PATH = + "../helm/flink-kubernetes-operator/crds/flinkstatesnapshots.flink.apache.org-v1.yml"; static KubernetesClient client; @@ -68,6 +76,65 @@ void upgradeCRDToYamlFlinkConfiguration() { assertThat(deployment.getSpec().getFlinkConfiguration().asFlatMap()).hasSize(5); } + /** + * Verifies that the validation constraints declared via CRD-generator annotations are enforced + * by the API server at admission time, rather than only by {@code + * DefaultValidator} during reconciliation. Covers {@code @Min(1)} on the JobManager/TaskManager + * replicas, {@code @Required} on {@code flinkVersion} and the ingress template, and + * {@code @Min(-1)} on the snapshot {@code backoffLimit}. + */ + @Test + void crdRejectsInvalidSpecValues() { + applyCRD(NEW_CRD_PATH); + + // A fully specified deployment satisfies all the schema constraints. + client.resource(validDeployment("valid-dep")).create(); + + // @Min(1) on JobManagerSpec.replicas + var jmReplicas = validDeployment("invalid-jm-replicas"); + jmReplicas.getSpec().getJobManager().setReplicas(0); + assertRejected(jmReplicas); + + // @Min(1) on TaskManagerSpec.replicas + var tmReplicas = validDeployment("invalid-tm-replicas"); + tmReplicas.getSpec().getTaskManager().setReplicas(0); + assertRejected(tmReplicas); + + // @Required on FlinkDeploymentSpec.flinkVersion + var noVersion = validDeployment("invalid-no-version"); + noVersion.getSpec().setFlinkVersion(null); + assertRejected(noVersion); + + // @Required on IngressSpec.template (an ingress without a template) + var noTemplate = validDeployment("invalid-ingress"); + noTemplate.getSpec().setIngress(new IngressSpec()); + assertRejected(noTemplate); + + // @Min(-1) on FlinkStateSnapshotSpec.backoffLimit (-1 is the unlimited-retries sentinel) + applyCRD(SNAPSHOT_CRD_PATH); + client.resource(validSnapshot("valid-snapshot")).create(); + var badBackoff = validSnapshot("invalid-backoff"); + badBackoff.getSpec().setBackoffLimit(-2); + assertRejected(badBackoff); + } + + private FlinkDeployment validDeployment(String name) { + var deployment = BaseTestUtils.buildApplicationCluster(); + deployment.getMetadata().setName(name); + deployment.getMetadata().setNamespace("default"); + return deployment; + } + + private FlinkStateSnapshot validSnapshot(String name) { + return BaseTestUtils.buildFlinkStateSnapshotSavepoint( + name, "default", "test-path", true, null); + } + + private void assertRejected(HasMetadata resource) { + assertThatThrownBy(() -> client.resource(resource).create()) + .isInstanceOf(KubernetesClientException.class); + } + private GenericKubernetesResource applyResource(String path) { try { GenericKubernetesResource d = diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java index 9525012c14..a72742a2d6 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java +++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java @@ -154,7 +154,9 @@ private Optional validateFlinkVersion(FlinkDeployment deployment) { var spec = deployment.getSpec(); var version = spec.getFlinkVersion(); if (version == null) { - return Optional.of("Flink Version must be defined."); + // Presence is enforced by the CRD schema (@Required on + // FlinkDeploymentSpec.flinkVersion); nothing further to validate here. + return Optional.empty(); } if (!FlinkVersion.isSupported(version)) { @@ -182,12 +184,11 @@ private Optional validateFlinkVersion(FlinkDeployment deployment) { } private Optional validateIngress(IngressSpec ingress, String name, String namespace) { - if (ingress == null) { + if (ingress == null || ingress.getTemplate() == null) { + // Template presence (when an ingress is configured) is enforced by the CRD schema + // (@Required on IngressSpec.template). return Optional.empty(); } - if (ingress.getTemplate() == null) { - return Optional.of("Ingress template must be defined"); - } try { IngressUtils.getIngressUrl(ingress.getTemplate(), name, namespace); } catch (ReconciliationException e) { @@ -294,9 +295,10 @@ private Optional validateJobSpec( var tmReplicasDefined = tm != null && tm.getReplicas() != null; - if (tmReplicasDefined && tm.getReplicas() < 1) { - return Optional.of("TaskManager replicas must be larger than 0"); - } else if (!tmReplicasDefined && job.getParallelism() < 1) { + // TaskManager replicas >= 1 is enforced by the CRD schema (@Min(1) on + // TaskManagerSpec.replicas). Only the parallelism fallback (used when replicas are not + // set) needs checking here, since parallelism defaults to 0 and is otherwise unconstrained. + if (!tmReplicasDefined && job.getParallelism() < 1) { return Optional.of("Job parallelism must be larger than 0"); } @@ -414,9 +416,8 @@ private Optional validateJmMemoryConfig(Configuration conf) { } private Optional validateJmReplicas(int replicas, Map confMap) { - if (replicas < 1) { - return Optional.of("JobManager replicas should not be configured less than one."); - } else if (replicas > 1 + // replicas >= 1 is enforced by the CRD schema (@Min(1) on JobManagerSpec.replicas). + if (replicas > 1 && !HighAvailabilityMode.isHighAvailabilityModeActivated( Configuration.fromMap(confMap))) { return Optional.of( @@ -442,8 +443,7 @@ private Optional validateTmSpec(TaskManagerSpec tmSpec, Map validateTmMemoryConfig(Configuration conf) { @@ -457,13 +457,6 @@ private Optional validateTmMemoryConfig(Configuration conf) { return Optional.empty(); } - private Optional validateTmReplicas(TaskManagerSpec tmSpec) { - if (tmSpec.getReplicas() != null && tmSpec.getReplicas() < 1) { - return Optional.of("TaskManager replicas should not be configured less than one."); - } - return Optional.empty(); - } - private Optional validateResources(String component, Resource resource) { if (resource == null) { return Optional.empty(); diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/validation/DefaultValidatorTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/validation/DefaultValidatorTest.java index ae28089bd1..e4f90aedbf 100644 --- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/validation/DefaultValidatorTest.java +++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/validation/DefaultValidatorTest.java @@ -39,7 +39,6 @@ import org.apache.flink.kubernetes.operator.api.spec.JobState; import org.apache.flink.kubernetes.operator.api.spec.KubernetesDeploymentMode; import org.apache.flink.kubernetes.operator.api.spec.SavepointSpec; -import org.apache.flink.kubernetes.operator.api.spec.TaskManagerSpec; import org.apache.flink.kubernetes.operator.api.spec.UpgradeMode; import org.apache.flink.kubernetes.operator.api.status.FlinkDeploymentReconciliationStatus; import org.apache.flink.kubernetes.operator.api.status.FlinkDeploymentStatus; @@ -100,13 +99,8 @@ public void testValidationWithoutDefaultConfig() { dep -> dep.getSpec().getJob().setParallelism(0), "Job parallelism must be larger than 0"); - testError( - dep -> { - var tmSpec = new TaskManagerSpec(); - tmSpec.setReplicas(0); - dep.getSpec().setTaskManager(tmSpec); - }, - "TaskManager replicas must be larger than 0"); + // TaskManager replicas < 1 is rejected by the CRD schema (@Min(1)) at admission, not by + // DefaultValidator; see FlinkConfigurationYamlSupportTest#crdRejectsInvalidSpecValues. testSuccess( dep -> { @@ -239,9 +233,9 @@ public void testValidationWithoutDefaultConfig() { Constants.CONFIG_FILE_LOG4J_NAME, "rootLogger.level = INFO"))); - testError( - dep -> dep.getSpec().setIngress(new IngressSpec()), - "Ingress template must be defined"); + // An ingress without a template is rejected by the CRD schema (@Required) at admission, + // not by DefaultValidator; see + // FlinkConfigurationYamlSupportTest#crdRejectsInvalidSpecValues. testError( dep -> @@ -279,9 +273,8 @@ public void testValidationWithoutDefaultConfig() { "true")), "HA must be enabled for rollback support."); - testError( - dep -> dep.getSpec().getJobManager().setReplicas(0), - "JobManager replicas should not be configured less than one."); + // JobManager replicas < 1 is rejected by the CRD schema (@Min(1)) at admission, not by + // DefaultValidator; see FlinkConfigurationYamlSupportTest#crdRejectsInvalidSpecValues. // Test resource validation testSuccessDeprecatedResource( @@ -577,7 +570,8 @@ public void testValidationWithoutDefaultConfig() { }, "Cannot switch from standalone kubernetes to native kubernetes cluster"); - testError(dep -> dep.getSpec().setFlinkVersion(null), "Flink Version must be defined."); + // A null flinkVersion is rejected by the CRD schema (@Required) at admission, not by + // DefaultValidator; see FlinkConfigurationYamlSupportTest#crdRejectsInvalidSpecValues. testError( dep -> dep.getSpec().setFlinkVersion(FlinkVersion.v1_14), diff --git a/helm/flink-kubernetes-operator/crds/flinkbluegreendeployments.flink.apache.org-v1.yml b/helm/flink-kubernetes-operator/crds/flinkbluegreendeployments.flink.apache.org-v1.yml index ae52bd0755..66fc393e59 100644 --- a/helm/flink-kubernetes-operator/crds/flinkbluegreendeployments.flink.apache.org-v1.yml +++ b/helm/flink-kubernetes-operator/crds/flinkbluegreendeployments.flink.apache.org-v1.yml @@ -53,6 +53,8 @@ spec: type: "string" type: "object" type: "array" + required: + - "template" type: "object" template: properties: @@ -176,6 +178,8 @@ spec: type: "string" type: "object" type: "array" + required: + - "template" type: "object" job: properties: @@ -3724,6 +3728,7 @@ spec: type: "object" type: "object" replicas: + minimum: 1.0 type: "integer" resource: properties: @@ -10793,6 +10798,7 @@ spec: type: "object" type: "object" replicas: + minimum: 1.0 type: "integer" resource: properties: @@ -10830,6 +10836,8 @@ spec: type: "object" type: "object" type: "object" + required: + - "flinkVersion" type: "object" type: "object" type: "object" diff --git a/helm/flink-kubernetes-operator/crds/flinkdeployments.flink.apache.org-v1.yml b/helm/flink-kubernetes-operator/crds/flinkdeployments.flink.apache.org-v1.yml index 426d822a08..68e87daa88 100644 --- a/helm/flink-kubernetes-operator/crds/flinkdeployments.flink.apache.org-v1.yml +++ b/helm/flink-kubernetes-operator/crds/flinkdeployments.flink.apache.org-v1.yml @@ -75,6 +75,8 @@ spec: type: "string" type: "object" type: "array" + required: + - "template" type: "object" job: properties: @@ -3623,6 +3625,7 @@ spec: type: "object" type: "object" replicas: + minimum: 1.0 type: "integer" resource: properties: @@ -10692,6 +10695,7 @@ spec: type: "object" type: "object" replicas: + minimum: 1.0 type: "integer" resource: properties: @@ -10729,6 +10733,8 @@ spec: type: "object" type: "object" type: "object" + required: + - "flinkVersion" type: "object" status: properties: diff --git a/helm/flink-kubernetes-operator/crds/flinkstatesnapshots.flink.apache.org-v1.yml b/helm/flink-kubernetes-operator/crds/flinkstatesnapshots.flink.apache.org-v1.yml index 06a05f7e2a..ff950706b0 100644 --- a/helm/flink-kubernetes-operator/crds/flinkstatesnapshots.flink.apache.org-v1.yml +++ b/helm/flink-kubernetes-operator/crds/flinkstatesnapshots.flink.apache.org-v1.yml @@ -33,6 +33,7 @@ spec: spec: properties: backoffLimit: + minimum: -1.0 type: "integer" checkpoint: type: "object"