refactor: add typed ClusterName to ValidatedCluster

Adds a top-level name: ClusterName field to ValidatedCluster (mirroring
trino-operator), constructed and validated in the validate step. The
per-file builders and build_rolegroup_config_map now take the whole
&ValidatedCluster, reading cluster.name and cluster.cluster_config.

product_version is intentionally not added: unlike Trino (integer
versions parsed to u16), HDFS uses semver product versions, so the image
keeps carrying the version string.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: maltesander

rust/operator-binary/src/controller/build/config_map.rs

@@ -85,15 +85,15 @@ pub fn build_rolegroup_config_map(
     let cluster_config = &cluster.cluster_config;
 
     let hdfs_site_xml = hdfs_site::build(
-        cluster_config,
+        cluster,
         cluster_info,
         merged_config,
         namenode_podrefs,
         journalnode_podrefs,
         config_overrides.hdfs_site_xml.clone(),
     );
     let core_site_xml = core_site::build(
-        cluster_config,
+        cluster,
         role,
         cluster_info,
         config_overrides.core_site_xml.clone(),

rust/operator-binary/src/controller/build/properties/core_site.rs

@@ -11,7 +11,7 @@ use crate::{
     config::CoreSiteConfigBuilder,
     controller::build::properties::resolved_overrides,
     crd::HdfsNodeRole,
-    hdfs_controller::ValidatedClusterConfig,
+    hdfs_controller::ValidatedCluster,
     security::kerberos::{self, KerberosConfig},
 };
 
@@ -24,20 +24,21 @@ pub enum Error {
 /// Renders `core-site.xml`: operator defaults + kerberos/OPA security config,
 /// with user `configOverrides` applied last.
 pub fn build(
-    cluster_config: &ValidatedClusterConfig,
+    cluster: &ValidatedCluster,
     role: HdfsNodeRole,
     cluster_info: &KubernetesClusterInfo,
     overrides: KeyValueConfigOverrides,
 ) -> Result<String, Error> {
+    let cluster_config = &cluster.cluster_config;
     let kerberos = KerberosConfig {
-        cluster_name: &cluster_config.name,
+        cluster_name: cluster.name.as_ref(),
         cluster_namespace: cluster_config.namespace.as_deref(),
         authentication_enabled: cluster_config.authentication_enabled,
         kerberos_enabled: cluster_config.kerberos_enabled,
         authorization_enabled: cluster_config.authorization_enabled,
     };
 
-    let mut core_site = CoreSiteConfigBuilder::new(cluster_config.name.clone());
+    let mut core_site = CoreSiteConfigBuilder::new(cluster.name.as_ref().to_owned());
     core_site
         .fs_default_fs()
         .ha_zookeeper_quorum()
@@ -67,13 +68,13 @@ pub fn build(
 mod tests {
     use super::*;
     use crate::controller::build::properties::test_support::{
-        cluster_info, config_overrides, validated_cluster_config,
+        cluster_info, config_overrides, validated_cluster,
     };
 
     #[test]
     fn renders_operator_defaults() {
         let xml = build(
-            &validated_cluster_config(),
+            &validated_cluster(),
             HdfsNodeRole::Name,
             &cluster_info(),
             config_overrides(&[]),
@@ -98,7 +99,7 @@ mod tests {
     #[test]
     fn user_overrides_win_over_defaults() {
         let xml = build(
-            &validated_cluster_config(),
+            &validated_cluster(),
             HdfsNodeRole::Name,
             &cluster_info(),
             config_overrides(&[("io.file.buffer.size", "65536")]),

rust/operator-binary/src/controller/build/properties/hdfs_site.rs

@@ -10,19 +10,20 @@ use crate::{
     config::HdfsSiteConfigBuilder,
     controller::build::properties::resolved_overrides,
     crd::{AnyNodeConfig, HdfsPodRef},
-    hdfs_controller::ValidatedClusterConfig,
+    hdfs_controller::ValidatedCluster,
 };
 
 /// Renders `hdfs-site.xml`: operator defaults, HA wiring derived from the pod
 /// refs, kerberos/OPA security config, with user `configOverrides` applied last.
 pub fn build(
-    cluster_config: &ValidatedClusterConfig,
+    cluster: &ValidatedCluster,
     cluster_info: &KubernetesClusterInfo,
     merged_config: &AnyNodeConfig,
     namenode_podrefs: &[HdfsPodRef],
     journalnode_podrefs: &[HdfsPodRef],
     overrides: KeyValueConfigOverrides,
 ) -> String {
+    let cluster_config = &cluster.cluster_config;
     // IMPORTANT: these folders must be under the volume mount point, otherwise they will not
     // be formatted by the namenode, or used by the other services.
     // See also: https://github.com/apache-spark-on-k8s/kubernetes-HDFS/commit/aef9586ecc8551ca0f0a468c3b917d8c38f494a0
@@ -34,7 +35,7 @@ pub fn build(
     // https://hadoop.apache.org/docs/stable/hadoop-project-dist/hadoop-hdfs/HDFSHighAvailabilityWithNFS.html
     // This caused a deadlock with no namenode becoming active during a startup after
     // HDFS was completely down for a while.
-    let mut hdfs_site = HdfsSiteConfigBuilder::new(cluster_config.name.clone());
+    let mut hdfs_site = HdfsSiteConfigBuilder::new(cluster.name.as_ref().to_owned());
     hdfs_site
         .dfs_namenode_name_dir()
         .dfs_datanode_data_dir(
@@ -114,7 +115,7 @@ mod tests {
     use super::*;
     use crate::{
         controller::build::properties::test_support::{
-            cluster_info, config_overrides, minimal_hdfs, validated_cluster_config,
+            cluster_info, config_overrides, minimal_hdfs, validated_cluster,
         },
         crd::{HdfsNodeRole, v1alpha1},
     };
@@ -129,7 +130,7 @@ mod tests {
     fn renders_operator_defaults() {
         let merged = namenode_merged_config(&minimal_hdfs());
         let xml = build(
-            &validated_cluster_config(),
+            &validated_cluster(),
             &cluster_info(),
             &merged,
             &[],
@@ -150,7 +151,7 @@ mod tests {
     fn user_overrides_win_over_defaults() {
         let merged = namenode_merged_config(&minimal_hdfs());
         let xml = build(
-            &validated_cluster_config(),
+            &validated_cluster(),
             &cluster_info(),
             &merged,
             &[],

rust/operator-binary/src/controller/build/properties/mod.rs

@@ -73,7 +73,9 @@ pub(crate) mod test_support {
         v2::config_overrides::KeyValueConfigOverrides,
     };
 
-    use crate::{crd::v1alpha1, hdfs_controller::ValidatedClusterConfig};
+    use crate::{
+        controller::validate::validate_cluster, crd::v1alpha1, hdfs_controller::ValidatedCluster,
+    };
 
     /// Builds a [`KeyValueConfigOverrides`] from `(key, value)` pairs for tests.
     pub fn config_overrides(pairs: &[(&str, &str)]) -> KeyValueConfigOverrides {
@@ -121,7 +123,8 @@ spec:
         }
     }
 
-    pub fn validated_cluster_config() -> ValidatedClusterConfig {
-        ValidatedClusterConfig::resolve(&minimal_hdfs(), None)
+    pub fn validated_cluster() -> ValidatedCluster {
+        validate_cluster(&minimal_hdfs(), "oci.example.org", None)
+            .expect("validate should succeed for the minimal fixture")
     }
 }

rust/operator-binary/src/controller/validate.rs

@@ -7,13 +7,15 @@
 //! validated via [`HdfsNodeRole::merged_config`], and the per-file
 //! `configOverrides` / `envOverrides` are merged here (role group wins).
 
-use std::collections::BTreeMap;
+use std::{collections::BTreeMap, str::FromStr};
 
 use snafu::{ResultExt, Snafu};
 use stackable_operator::{
     commons::product_image_selection,
     config::merge::Merge,
+    kube::ResourceExt,
     role_utils::{GenericRoleConfig, JavaCommonConfig, Role},
+    v2::types::operator::ClusterName,
 };
 use strum::IntoEnumIterator;
 
@@ -33,6 +35,11 @@ pub enum Error {
         source: product_image_selection::Error,
     },
 
+    #[snafu(display("invalid cluster name"))]
+    InvalidClusterName {
+        source: stackable_operator::v2::macros::attributed_string_type::Error,
+    },
+
     #[snafu(display("failed to resolve and merge config for role and role group"))]
     FailedToResolveConfig { source: crate::crd::Error },
 }
@@ -97,6 +104,7 @@ pub fn validate_cluster(
     }
 
     Ok(ValidatedCluster {
+        name: ClusterName::from_str(&hdfs.name_any()).context(InvalidClusterNameSnafu)?,
         image: resolved_product_image,
         cluster_config: ValidatedClusterConfig::resolve(hdfs, hdfs_opa_config),
         role_groups,

rust/operator-binary/src/hdfs_controller.rs

@@ -38,6 +38,7 @@ use stackable_operator::{
         rollout::check_statefulset_rollout_complete,
     },
     utils::cluster_info::KubernetesClusterInfo,
+    v2::types::operator::ClusterName,
 };
 use strum::{EnumDiscriminants, IntoEnumIterator, IntoStaticStr};
 
@@ -70,6 +71,8 @@ pub const CONTAINER_IMAGE_BASE_NAME: &str = "hadoop";
 /// the controller.
 #[derive(Clone, Debug)]
 pub struct ValidatedCluster {
+    /// The logical (and Kubernetes object) name of the cluster.
+    pub name: ClusterName,
     pub image: ResolvedProductImage,
     pub cluster_config: ValidatedClusterConfig,
     pub role_groups: BTreeMap<HdfsNodeRole, BTreeMap<String, ValidatedRoleGroupConfig>>,
@@ -81,8 +84,6 @@ pub struct ValidatedCluster {
 /// the same `HdfsCluster` predicates used previously, just resolved up-front.
 #[derive(Clone, Debug)]
 pub struct ValidatedClusterConfig {
-    /// The logical (and Kubernetes object) name of the cluster.
-    pub name: String,
     /// The cluster namespace, used to build kerberos principals.
     pub namespace: Option<String>,
     pub dfs_replication: u8,
@@ -100,7 +101,6 @@ impl ValidatedClusterConfig {
         authorization: Option<HdfsOpaConfig>,
     ) -> ValidatedClusterConfig {
         ValidatedClusterConfig {
-            name: hdfs.name_any(),
             namespace: hdfs.namespace(),
             dfs_replication: hdfs.spec.cluster_config.dfs_replication,
             https_enabled: hdfs.has_https_enabled(),