Skip to content

[FLINK-31849] Adds basic CRD validations #1144

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
vsantwana wants to merge 1 commit into
base: main
Choose a base branch
from
Open

[FLINK-31849] Adds basic CRD validations #1144

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion ...-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/FlinkDeploymentSpec.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand Down
2 changes: 2 additions & 0 deletions ...i/src/main/java/org/apache/flink/kubernetes/operator/api/spec/FlinkStateSnapshotSpec.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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)

@Dennis-Mircea Dennis-Mircea Jul 2, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This doesn't look as a necessary validation constraint.

private int backoffLimit = -1;

public boolean isSavepoint() {
Expand Down
3 changes: 2 additions & 1 deletion ...operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/IngressSpec.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down
2 changes: 2 additions & 0 deletions ...rator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/JobManagerSpec.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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. */
Expand Down
2 changes: 2 additions & 0 deletions ...ator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/TaskManagerSpec.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -57,6 +58,7 @@ public class TaskManagerSpec implements Diffable<TaskManagerSpec> {
/** 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. */
Expand Down
67 changes: 67 additions & 0 deletions ...test/java/org/apache/flink/kubernetes/operator/api/FlinkConfigurationYamlSupportTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,19 +17,25 @@

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;

import java.io.File;
import java.io.IOException;

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;

@EnableKubeAPIServer
class FlinkConfigurationYamlSupportTest {
Expand All @@ -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;

Expand Down Expand Up @@ -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 =
Expand Down
33 changes: 13 additions & 20 deletions ...rator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -154,7 +154,9 @@ private Optional<String> 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();

@Dennis-Mircea Dennis-Mircea Jul 2, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The new validation is admission-only and does not cover the reconcile cycle. The new added constraints are enforced only at admission, and only if the updated CRD is installed, which is a manual step (Helm does not auto-upgrade crds/). DefaultValidator, however, runs on every reconcile as well, so removing these checks drops the runtime validation checks: after an operator bump where the CRD has not been replaced yet, an invalid spec reaching reconcile is no longer caught and hits deploy logic with a bad value.

}

if (!FlinkVersion.isSupported(version)) {
Expand Down Expand Up @@ -182,12 +184,11 @@ private Optional<String> validateFlinkVersion(FlinkDeployment deployment) {
}

private Optional<String> 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) {
Expand Down Expand Up @@ -294,9 +295,10 @@ private Optional<String> 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");
}

Expand Down Expand Up @@ -414,9 +416,8 @@ private Optional<String> validateJmMemoryConfig(Configuration conf) {
}

private Optional<String> validateJmReplicas(int replicas, Map<String, String> 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(
Expand All @@ -442,8 +443,7 @@ private Optional<String> validateTmSpec(TaskManagerSpec tmSpec, Map<String, Stri
return firstPresent(
tmMemoryConfigValidation,
validateResources("TaskManager", tmSpec.getResource()),
validateResourceRequirements("TaskManager", tmSpec.getResources()),
validateTmReplicas(tmSpec));
validateResourceRequirements("TaskManager", tmSpec.getResources()));
}

private Optional<String> validateTmMemoryConfig(Configuration conf) {
Expand All @@ -457,13 +457,6 @@ private Optional<String> validateTmMemoryConfig(Configuration conf) {
return Optional.empty();
}

private Optional<String> 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<String> validateResources(String component, Resource resource) {
if (resource == null) {
return Optional.empty();
Expand Down
24 changes: 9 additions & 15 deletions ...r/src/test/java/org/apache/flink/kubernetes/operator/validation/DefaultValidatorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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 -> {
Expand Down Expand Up @@ -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 ->
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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),
Expand Down
8 changes: 8 additions & 0 deletions helm/flink-kubernetes-operator/crds/flinkbluegreendeployments.flink.apache.org-v1.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,8 @@ spec:
type: "string"
type: "object"
type: "array"
required:
- "template"
type: "object"
template:
properties:
Expand Down Expand Up @@ -176,6 +178,8 @@ spec:
type: "string"
type: "object"
type: "array"
required:
- "template"
type: "object"
job:
properties:
Expand Down Expand Up @@ -3724,6 +3728,7 @@ spec:
type: "object"
type: "object"
replicas:
minimum: 1.0
type: "integer"
resource:
properties:
Expand Down Expand Up @@ -10793,6 +10798,7 @@ spec:
type: "object"
type: "object"
replicas:
minimum: 1.0
type: "integer"
resource:
properties:
Expand Down Expand Up @@ -10830,6 +10836,8 @@ spec:
type: "object"
type: "object"
type: "object"
required:
- "flinkVersion"
type: "object"
type: "object"
type: "object"
Expand Down
6 changes: 6 additions & 0 deletions helm/flink-kubernetes-operator/crds/flinkdeployments.flink.apache.org-v1.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,6 +75,8 @@ spec:
type: "string"
type: "object"
type: "array"
required:
- "template"
type: "object"
job:
properties:
Expand Down Expand Up @@ -3623,6 +3625,7 @@ spec:
type: "object"
type: "object"
replicas:
minimum: 1.0
type: "integer"
resource:
properties:
Expand Down Expand Up @@ -10692,6 +10695,7 @@ spec:
type: "object"
type: "object"
replicas:
minimum: 1.0
type: "integer"
resource:
properties:
Expand Down Expand Up @@ -10729,6 +10733,8 @@ spec:
type: "object"
type: "object"
type: "object"
required:
- "flinkVersion"
type: "object"
status:
properties:
Expand Down
Loading