fix(helm): synchronize FlinkCluster CRD schema (#1027)

Refresh the Helm CRD from the canonical generated schema so Helm installs no longer persist a CANONICAL savepoint format by default. Add a CI consistency gate and document upgrade behavior for existing resources.

Author: regadas

Makefile

@@ -52,6 +52,24 @@ generate: controller-gen ## Generate code containing DeepCopy, DeepCopyInto, and
 generate-crd-docs: crd-ref-docs ## Generate CRD documentation to docs/crd.md
 	$(CRD_REF_DOCS) --source-path=./apis/flinkcluster/v1beta1 --config=docs/config.yaml --renderer=markdown --output-path=docs/crd.md
 
+.PHONY: verify-helm-crd-schema
+verify-helm-crd-schema: kustomize ## Verify the Helm CRD schema matches the generated CRD.
+	@set -euo pipefail; \
+	tmp_dir=$$(mktemp -d); \
+	trap 'rm -rf "$$tmp_dir"' EXIT; \
+	$(KUSTOMIZE) build config/default | \
+		yq -o=json -I=2 'select(.kind == "CustomResourceDefinition" and .metadata.name == "flinkclusters.flinkoperator.k8s.io") | .spec.versions[] | select(.name == "v1beta1") | .schema.openAPIV3Schema | sort_keys(..)' - > "$$tmp_dir/generated.json"; \
+	sed -E 's/\{\{[^}]+\}\}/helm-value/g' helm-chart/flink-operator/templates/flink-cluster-crd.yaml | \
+		yq -o=json -I=2 '.spec.versions[] | select(.name == "v1beta1") | .schema.openAPIV3Schema | sort_keys(..)' - > "$$tmp_dir/helm.json"; \
+	grep -q '^{' "$$tmp_dir/generated.json" && grep -q '^{' "$$tmp_dir/helm.json" || { \
+		echo "Failed to extract the v1beta1 OpenAPI schema for comparison." >&2; \
+		exit 1; \
+	}; \
+	diff -u "$$tmp_dir/generated.json" "$$tmp_dir/helm.json" || { \
+		echo "Helm CRD schema is out of sync. Run 'cd helm-chart/flink-operator && ./update_template.sh' and commit the refreshed CRD template." >&2; \
+		exit 1; \
+	}
+
 tidy: ## Run go mod tidy
 	go mod tidy
 
@@ -63,6 +81,7 @@ vet: ## Run go vet against code.
 
 .PHONY: test
 test: manifests generate fmt vet tidy kustomize envtest ## Run tests.
+	$(MAKE) verify-helm-crd-schema
 	rm -rf config/test && mkdir -p config/test/crd
 	$(KUSTOMIZE) build config/crd > config/test/crd/flinkoperator.k8s.io_flinkclusters.yaml
 	KUBEBUILDER_ASSETS=$(shell $(ENVTEST) use $(ENVTEST_K8S_VERSION) --bin-dir $(LOCALBIN) -p path) go test ./... -coverprofile cover.out

docs/savepoints_guide.md

@@ -245,8 +245,12 @@ savepoints in GCS.
 ## Savepoint format
 
 Flink supports two savepoint formats: `CANONICAL` and `NATIVE`. You can control which format is used by setting the
-`savepointFormatType` property in the job spec. If not specified, Flink uses the canonical format by default. This option
-requires Flink 1.15 or later; for older or unparseable Flink versions, the operator omits the format parameter.
+`savepointFormatType` property in the job spec. If it is omitted, the operator omits the REST format parameter and
+`status.savepoint.formatType`, allowing Flink to choose its own default (currently canonical). The operator also omits
+this parameter for Flink versions before 1.15 or when the Flink version cannot be parsed.
+
+Removing the CRD default is not retroactive. Clusters created while an older CRD defaulted this field may retain an
+explicit `CANONICAL` value in their stored spec until the field is removed or changed.
 
 For more details on the differences between the two formats and when to use each, see the
 [Flink documentation on savepoint formats](https://nightlies.apache.org/flink/flink-docs-stable/docs/ops/state/savepoints/#savepoint-format).