refactor: introduce ValidatedClusterConfig

Author: maltesander

rust/operator-binary/src/controller.rs

@@ -209,44 +209,34 @@ impl ReconcilerError for Error {
     }
 }
 
+pub type RoleGroupName = String;
+
 /// The validated cluster. Carries everything the build steps need, resolved once
 /// here so downstream code never re-derives it or touches the raw spec.
 ///
 /// The cluster identity (`name`, `namespace`, `uid`) is captured here so that owner
 /// references for child objects can be built straight from this struct (via its
 /// [`Resource`] impl) without threading the raw [`v1alpha1::KafkaCluster`] around.
 /// This mirrors the hive-/opensearch-operator's `ValidatedCluster`.
-pub struct ValidatedKafkaCluster {
+pub struct ValidatedCluster {
     /// `ObjectMeta` carrying `name`, `namespace` and `uid`, so this struct can act as the
     /// owner [`Resource`] for child objects.
     metadata: ObjectMeta,
     pub name: ClusterName,
     pub namespace: NamespaceName,
     pub image: ResolvedProductImage,
-    pub kafka_security: KafkaTlsSecurity,
-    // DESIGN DECISION: the dereferenced authorization config is folded into the
-    // validated cluster (read from here downstream). The other dereferenced input,
-    // the authentication classes, is intentionally NOT stored: it is fully consumed
-    // here to build `kafka_security`. Alternative: also store the resolved auth
-    // classes — rejected because nothing downstream needs them beyond kafka_security.
-    pub authorization_config: Option<KafkaAuthorizationConfig>,
-    pub role_groups: BTreeMap<KafkaRole, BTreeMap<String, ValidatedRoleGroupConfig>>,
-    pub pod_descriptors: Vec<KafkaPodDescriptor>,
-    pub metadata_manager: MetadataManager,
+    pub cluster_config: ValidatedClusterConfig,
+    pub role_group_configs: BTreeMap<KafkaRole, BTreeMap<RoleGroupName, ValidatedRoleGroupConfig>>,
 }
 
-impl ValidatedKafkaCluster {
-    #[allow(clippy::too_many_arguments)]
+impl ValidatedCluster {
     pub fn new(
         name: ClusterName,
         namespace: NamespaceName,
         uid: Uid,
         image: ResolvedProductImage,
-        kafka_security: KafkaTlsSecurity,
-        authorization_config: Option<KafkaAuthorizationConfig>,
-        role_groups: BTreeMap<KafkaRole, BTreeMap<String, ValidatedRoleGroupConfig>>,
-        pod_descriptors: Vec<KafkaPodDescriptor>,
-        metadata_manager: MetadataManager,
+        cluster_config: ValidatedClusterConfig,
+        role_group_configs: BTreeMap<KafkaRole, BTreeMap<RoleGroupName, ValidatedRoleGroupConfig>>,
     ) -> Self {
         Self {
             metadata: ObjectMeta {
@@ -258,18 +248,26 @@ impl ValidatedKafkaCluster {
             name,
             namespace,
             image,
-            kafka_security,
-            authorization_config,
-            role_groups,
-            pod_descriptors,
-            metadata_manager,
+            cluster_config,
+            role_group_configs,
         }
     }
 }
 
-/// Lets [`ValidatedKafkaCluster`] act as the owner [`Resource`] for child objects, so owner
+/// Cluster-wide settings resolved during validation and dereferencing.
+///
+/// Everything the build steps need is resolved here so they never have to read the
+/// raw [`v1alpha1::KafkaCluster`] spec.
+pub struct ValidatedClusterConfig {
+    pub kafka_security: KafkaTlsSecurity,
+    pub authorization_config: Option<KafkaAuthorizationConfig>,
+    pub pod_descriptors: Vec<KafkaPodDescriptor>,
+    pub metadata_manager: MetadataManager,
+}
+
+/// Lets [`ValidatedCluster`] act as the owner [`Resource`] for child objects, so owner
 /// references are built from it (via the captured `metadata`) rather than the raw CR.
-impl Resource for ValidatedKafkaCluster {
+impl Resource for ValidatedCluster {
     type DynamicType = <v1alpha1::KafkaCluster as Resource>::DynamicType;
     type Scope = <v1alpha1::KafkaCluster as Resource>::Scope;
 
@@ -348,10 +346,10 @@ pub async fn reconcile_kafka(
     .context(CreateClusterResourcesSnafu)?;
 
     tracing::debug!(
-        kerberos_enabled = validated_cluster.kafka_security.has_kerberos_enabled(),
-        kerberos_secret_class = ?validated_cluster.kafka_security.kerberos_secret_class(),
-        tls_enabled = validated_cluster.kafka_security.tls_enabled(),
-        tls_client_authentication_class = ?validated_cluster.kafka_security.tls_client_authentication_class(),
+        kerberos_enabled = validated_cluster.cluster_config.kafka_security.has_kerberos_enabled(),
+        kerberos_secret_class = ?validated_cluster.cluster_config.kafka_security.kerberos_secret_class(),
+        tls_enabled = validated_cluster.cluster_config.kafka_security.tls_enabled(),
+        tls_client_authentication_class = ?validated_cluster.cluster_config.kafka_security.tls_client_authentication_class(),
         "The following security settings are used"
     );
 
@@ -377,15 +375,15 @@ pub async fn reconcile_kafka(
 
     let mut bootstrap_listeners = Vec::<listener::v1alpha1::Listener>::new();
 
-    for (kafka_role, rg_map) in &validated_cluster.role_groups {
+    for (kafka_role, rg_map) in &validated_cluster.role_group_configs {
         for (rolegroup_name, validated_rg) in rg_map {
             let rolegroup_ref = kafka.rolegroup_ref(kafka_role, rolegroup_name);
 
             let rg_headless_service = build_rolegroup_headless_service(
                 &validated_cluster,
                 &validated_cluster.image,
                 &rolegroup_ref,
-                &validated_cluster.kafka_security,
+                &validated_cluster.cluster_config.kafka_security,
             )
             .context(BuildServiceSnafu)?;
 
@@ -398,7 +396,7 @@ pub async fn reconcile_kafka(
 
             let kafka_listeners = get_kafka_listener_config(
                 kafka,
-                &validated_cluster.kafka_security,
+                &validated_cluster.cluster_config.kafka_security,
                 &rolegroup_ref,
                 &client.kubernetes_cluster_info,
             )

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

@@ -8,7 +8,7 @@ use stackable_operator::{
 };
 
 use crate::{
-    controller::{KAFKA_CONTROLLER_NAME, ValidatedKafkaCluster, ValidatedRoleGroupConfig},
+    controller::{KAFKA_CONTROLLER_NAME, ValidatedCluster, ValidatedRoleGroupConfig},
     crd::{
         JVM_SECURITY_PROPERTIES_FILE, MetadataManager, STACKABLE_LISTENER_BOOTSTRAP_DIR,
         STACKABLE_LISTENER_BROKER_DIR,
@@ -69,32 +69,33 @@ pub enum Error {
 /// The rolegroup [`ConfigMap`] configures the rolegroup based on the configuration given by the administrator
 pub fn build_rolegroup_config_map(
     kafka: &v1alpha1::KafkaCluster,
-    validated_cluster: &ValidatedKafkaCluster,
+    validated_cluster: &ValidatedCluster,
     rolegroup: &RoleGroupRef<v1alpha1::KafkaCluster>,
     validated_rg: &ValidatedRoleGroupConfig,
     listener_config: &KafkaListenerConfig,
 ) -> Result<ConfigMap, Error> {
-    let kafka_security = &validated_cluster.kafka_security;
+    let kafka_security = &validated_cluster.cluster_config.kafka_security;
     let resolved_product_image = &validated_cluster.image;
     let kafka_config_file_name = validated_rg.merged_config.config_file_name();
     let config_overrides = validated_rg.config_file_overrides.clone();
 
     let opa_connect = validated_cluster
+        .cluster_config
         .authorization_config
         .as_ref()
         .map(|auth_config| auth_config.opa_connect.clone());
 
-    let kraft_mode = validated_cluster.metadata_manager == MetadataManager::KRaft;
+    let kraft_mode = validated_cluster.cluster_config.metadata_manager == MetadataManager::KRaft;
 
-    if kraft_mode && validated_cluster.pod_descriptors.is_empty() {
+    if kraft_mode && validated_cluster.cluster_config.pod_descriptors.is_empty() {
         return NoKraftControllersFoundSnafu.fail();
     }
 
     let kafka_config = match &validated_rg.merged_config {
         AnyConfig::Broker(_) => crate::controller::build::properties::broker_properties::build(
             kafka_security,
             listener_config,
-            &validated_cluster.pod_descriptors,
+            &validated_cluster.cluster_config.pod_descriptors,
             opa_connect.as_deref(),
             kraft_mode,
             kafka
@@ -108,7 +109,7 @@ pub fn build_rolegroup_config_map(
             crate::controller::build::properties::controller_properties::build(
                 kafka_security,
                 listener_config,
-                &validated_cluster.pod_descriptors,
+                &validated_cluster.cluster_config.pod_descriptors,
                 kraft_mode,
                 config_overrides,
             )

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

@@ -9,7 +9,7 @@ use stackable_operator::{
 };
 
 use crate::{
-    controller::{KAFKA_CONTROLLER_NAME, ValidatedKafkaCluster},
+    controller::{KAFKA_CONTROLLER_NAME, ValidatedCluster},
     crd::{role::KafkaRole, v1alpha1},
     utils::build_recommended_labels,
 };
@@ -42,10 +42,10 @@ pub enum Error {
 /// Build a discovery [`ConfigMap`] containing information about how to connect to a certain
 /// [`v1alpha1::KafkaCluster`].
 pub fn build_discovery_configmap(
-    validated_cluster: &ValidatedKafkaCluster,
+    validated_cluster: &ValidatedCluster,
     listeners: &[listener::v1alpha1::Listener],
 ) -> Result<ConfigMap, Error> {
-    let kafka_security = &validated_cluster.kafka_security;
+    let kafka_security = &validated_cluster.cluster_config.kafka_security;
     let resolved_product_image = &validated_cluster.image;
 
     let port_name = if kafka_security.has_kerberos_enabled() {
@@ -87,7 +87,7 @@ pub fn build_discovery_configmap(
 
 /// An [`ObjectRef`] to the owning cluster, built from the validated identity — used only for
 /// error context.
-fn cluster_object_ref(cluster: &ValidatedKafkaCluster) -> ObjectRef<v1alpha1::KafkaCluster> {
+fn cluster_object_ref(cluster: &ValidatedCluster) -> ObjectRef<v1alpha1::KafkaCluster> {
     ObjectRef::new(cluster.name.as_ref()).within(cluster.namespace.as_ref())
 }
 

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

@@ -1,7 +1,7 @@
 //! The validate step in the KafkaCluster controller.
 //!
 //! Synchronously validates inputs that don't require a Kubernetes client. Produces
-//! [`ValidatedKafkaCluster`], consumed by the rest of `reconcile_kafka`.
+//! [`ValidatedCluster`], consumed by the rest of `reconcile_kafka`.
 
 use std::{collections::BTreeMap, str::FromStr};
 
@@ -22,7 +22,8 @@ use stackable_operator::{
 
 use crate::{
     controller::{
-        ValidatedKafkaCluster, ValidatedRoleGroupConfig, dereference::DereferencedObjects,
+        RoleGroupName, ValidatedCluster, ValidatedClusterConfig, ValidatedRoleGroupConfig,
+        dereference::DereferencedObjects,
     },
     crd::{
         self, CONTAINER_IMAGE_BASE_NAME,
@@ -87,7 +88,7 @@ pub fn validate(
     kafka: &v1alpha1::KafkaCluster,
     dereferenced_objects: DereferencedObjects,
     operator_environment: &OperatorEnvironmentOptions,
-) -> Result<ValidatedKafkaCluster> {
+) -> Result<ValidatedCluster> {
     let image = kafka
         .spec
         .image
@@ -115,13 +116,10 @@ pub fn validate(
         .validate_authentication_methods()
         .context(FailedToValidateAuthenticationMethodSnafu)?;
 
-    // DESIGN DECISION: build the per-rolegroup config (merged config + resolved overrides)
-    // here, so reconcile reads a fully-typed ValidatedKafkaCluster instead of re-deriving
-    // merged_config in the loop and threading a product-config HashMap. Alternative: keep
-    // deriving merged_config in the reconcile loop — rejected; validation is the right place
-    // to prove every rolegroup resolves before any resource is built.
-    let mut role_groups: BTreeMap<KafkaRole, BTreeMap<String, ValidatedRoleGroupConfig>> =
-        BTreeMap::new();
+    let mut role_group_configs: BTreeMap<
+        KafkaRole,
+        BTreeMap<RoleGroupName, ValidatedRoleGroupConfig>,
+    > = BTreeMap::new();
 
     // Brokers always exist.
     let broker_role = kafka
@@ -131,7 +129,7 @@ pub fn validate(
             role: KafkaRole::Broker,
         })?;
 
-    let mut broker_groups: BTreeMap<String, ValidatedRoleGroupConfig> = BTreeMap::new();
+    let mut broker_groups: BTreeMap<RoleGroupName, ValidatedRoleGroupConfig> = BTreeMap::new();
     for rolegroup_name in broker_role.role_groups.keys() {
         let merged_config = KafkaRole::Broker
             .merged_config(kafka, rolegroup_name)
@@ -148,7 +146,7 @@ pub fn validate(
             },
         );
     }
-    role_groups.insert(KafkaRole::Broker, broker_groups);
+    role_group_configs.insert(KafkaRole::Broker, broker_groups);
 
     // We need this guard because controller_role() returns an error if controllers is None,
     // which would stop reconciliation for ZooKeeper-mode clusters.
@@ -160,7 +158,8 @@ pub fn validate(
                 role: KafkaRole::Controller,
             })?;
 
-        let mut controller_groups: BTreeMap<String, ValidatedRoleGroupConfig> = BTreeMap::new();
+        let mut controller_groups: BTreeMap<RoleGroupName, ValidatedRoleGroupConfig> =
+            BTreeMap::new();
         for rolegroup_name in controller_role.role_groups.keys() {
             let merged_config = KafkaRole::Controller
                 .merged_config(kafka, rolegroup_name)
@@ -177,7 +176,7 @@ pub fn validate(
                 },
             );
         }
-        role_groups.insert(KafkaRole::Controller, controller_groups);
+        role_group_configs.insert(KafkaRole::Controller, controller_groups);
     }
 
     let pod_descriptors = kafka
@@ -197,16 +196,18 @@ pub fn validate(
         .context(InvalidNamespaceSnafu)?;
     let uid = Uid::from_str(&kafka.uid().context(ObjectHasNoUidSnafu)?).context(InvalidUidSnafu)?;
 
-    Ok(ValidatedKafkaCluster::new(
+    Ok(ValidatedCluster::new(
         name,
         namespace,
         uid,
         image,
-        kafka_security,
-        dereferenced_objects.authorization_config,
-        role_groups,
-        pod_descriptors,
-        metadata_manager,
+        ValidatedClusterConfig {
+            kafka_security,
+            authorization_config: dereferenced_objects.authorization_config,
+            pod_descriptors,
+            metadata_manager,
+        },
+        role_group_configs,
     ))
 }
 

rust/operator-binary/src/resource/listener.rs

@@ -4,7 +4,7 @@ use stackable_operator::{
 };
 
 use crate::{
-    controller::{KAFKA_CONTROLLER_NAME, ValidatedKafkaCluster},
+    controller::{KAFKA_CONTROLLER_NAME, ValidatedCluster},
     crd::{role::broker::BrokerConfig, security::KafkaTlsSecurity, v1alpha1},
     utils::build_recommended_labels,
 };
@@ -27,11 +27,11 @@ pub enum Error {
 // TODO (@NickLarsenNZ): Move shared functionality to stackable-operator
 pub fn build_broker_rolegroup_bootstrap_listener(
     kafka: &v1alpha1::KafkaCluster,
-    validated_cluster: &ValidatedKafkaCluster,
+    validated_cluster: &ValidatedCluster,
     rolegroup: &RoleGroupRef<v1alpha1::KafkaCluster>,
     merged_config: &BrokerConfig,
 ) -> Result<listener::v1alpha1::Listener, Error> {
-    let kafka_security = &validated_cluster.kafka_security;
+    let kafka_security = &validated_cluster.cluster_config.kafka_security;
     let resolved_product_image = &validated_cluster.image;
 
     Ok(listener::v1alpha1::Listener {

rust/operator-binary/src/resource/service.rs

@@ -8,7 +8,7 @@ use stackable_operator::{
 };
 
 use crate::{
-    controller::{KAFKA_CONTROLLER_NAME, ValidatedKafkaCluster},
+    controller::{KAFKA_CONTROLLER_NAME, ValidatedCluster},
     crd::{APP_NAME, METRICS_PORT, METRICS_PORT_NAME, security::KafkaTlsSecurity, v1alpha1},
     utils::build_recommended_labels,
 };
@@ -35,7 +35,7 @@ pub enum Error {
 ///
 /// This is mostly useful for internal communication between peers, or for clients that perform client-side load balancing.
 pub fn build_rolegroup_headless_service(
-    validated_cluster: &ValidatedKafkaCluster,
+    validated_cluster: &ValidatedCluster,
     resolved_product_image: &ResolvedProductImage,
     rolegroup: &RoleGroupRef<v1alpha1::KafkaCluster>,
     kafka_security: &KafkaTlsSecurity,
@@ -77,7 +77,7 @@ pub fn build_rolegroup_headless_service(
 
 /// The rolegroup metrics [`Service`] is a service that exposes metrics and a prometheus scraping label
 pub fn build_rolegroup_metrics_service(
-    validated_cluster: &ValidatedKafkaCluster,
+    validated_cluster: &ValidatedCluster,
     resolved_product_image: &ResolvedProductImage,
     rolegroup: &RoleGroupRef<v1alpha1::KafkaCluster>,
 ) -> Result<Service, Error> {