move resolution of metadata_manager and pod descriptors to validate stage

Author: adwk67

rust/operator-binary/src/controller.rs

@@ -22,7 +22,6 @@ use stackable_operator::{
         compute_conditions, operations::ClusterOperationsConditionBuilder,
         statefulset::StatefulSetConditionBuilder,
     },
-    utils::cluster_info::KubernetesClusterInfo,
 };
 use strum::{EnumDiscriminants, IntoStaticStr};
 
@@ -32,7 +31,7 @@ mod validate;
 
 use crate::{
     crd::{
-        self, APP_NAME, KafkaClusterStatus, OPERATOR_NAME,
+        self, APP_NAME, KafkaClusterStatus, KafkaPodDescriptor, MetadataManager, OPERATOR_NAME,
         authorization::KafkaAuthorizationConfig,
         listener::get_kafka_listener_config,
         role::{AnyConfig, KafkaRole},
@@ -219,7 +218,8 @@ pub struct ValidatedKafkaCluster {
     // classes — rejected because nothing downstream needs them beyond kafka_security.
     pub authorization_config: Option<KafkaAuthorizationConfig>,
     pub role_groups: BTreeMap<KafkaRole, BTreeMap<String, ValidatedRoleGroupConfig>>,
-    pub kubernetes_cluster_info: KubernetesClusterInfo,
+    pub pod_descriptors: Vec<KafkaPodDescriptor>,
+    pub metadata_manager: MetadataManager,
 }
 
 pub struct ValidatedRoleGroupConfig {

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

@@ -58,17 +58,14 @@ pub enum Error {
         rolegroup: RoleGroupRef<v1alpha1::KafkaCluster>,
     },
 
-    #[snafu(display("failed to build properties for {rolegroup}"))]
-    BuildProperties {
-        source: crate::controller::build::properties::Error,
-        rolegroup: RoleGroupRef<v1alpha1::KafkaCluster>,
-    },
-
     #[snafu(display("failed to build jaas configuration file for {rolegroup}"))]
     BuildJaasConfig { rolegroup: String },
 
     #[snafu(display("failed to build pod descriptors"))]
     BuildPodDescriptors { source: crate::crd::Error },
+
+    #[snafu(display("no Kraft controllers found to build"))]
+    NoKraftControllersFound,
 }
 
 /// The rolegroup [`ConfigMap`] configures the rolegroup based on the configuration given by the administrator
@@ -89,25 +86,19 @@ pub fn build_rolegroup_config_map(
         .as_ref()
         .map(|auth_config| auth_config.opa_connect.clone());
 
-    let pod_descriptors = &kafka
-        .pod_descriptors(
-            None,
-            &validated_cluster.kubernetes_cluster_info,
-            validated_cluster.kafka_security.client_port(),
-        )
-        .context(BuildPodDescriptorsSnafu)?;
+    let kraft_mode = validated_cluster.metadata_manager == MetadataManager::KRaft;
 
-    let metadata_manager = kafka
-        .effective_metadata_manager()
-        .context(InvalidMetadataManagerSnafu)?;
+    if kraft_mode && validated_cluster.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,
-            pod_descriptors,
+            &validated_cluster.pod_descriptors,
             opa_connect.as_deref(),
-            metadata_manager == MetadataManager::KRaft,
+            kraft_mode,
             kafka
                 .spec
                 .cluster_config
@@ -119,15 +110,12 @@ pub fn build_rolegroup_config_map(
             crate::controller::build::properties::controller_properties::build(
                 kafka_security,
                 listener_config,
-                pod_descriptors,
-                metadata_manager == MetadataManager::KRaft,
+                &validated_cluster.pod_descriptors,
+                kraft_mode,
                 config_overrides,
             )
         }
-    }
-    .with_context(|_| BuildPropertiesSnafu {
-        rolegroup: rolegroup.clone(),
-    })?;
+    };
 
     let kafka_config = kafka_config
         .into_iter()

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

@@ -1,8 +1,6 @@
 use std::collections::BTreeMap;
 
-use snafu::OptionExt;
-
-use super::{Error, NoKraftControllersFoundSnafu, kraft_controllers};
+use super::kraft_controllers;
 use crate::{
     crd::{
         KafkaPodDescriptor,
@@ -25,7 +23,7 @@ pub fn build(
     kraft_mode: bool,
     disable_broker_id_generation: bool,
     overrides: BTreeMap<String, String>,
-) -> Result<BTreeMap<String, String>, Error> {
+) -> BTreeMap<String, String> {
     let kraft_controllers = kraft_controllers(pod_descriptors);
 
     let mut result = BTreeMap::from([
@@ -49,7 +47,7 @@ pub fn build(
     ]);
 
     if kraft_mode {
-        let kraft_controllers = kraft_controllers.context(NoKraftControllersFoundSnafu)?;
+        let kraft_controllers = kraft_controllers.join(",");
 
         // Running in KRaft mode
         result.extend([
@@ -114,5 +112,5 @@ pub fn build(
     result.extend(graceful_shutdown_config_properties());
     result.extend(overrides);
 
-    Ok(result)
+    result
 }

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

@@ -1,8 +1,6 @@
 use std::collections::BTreeMap;
 
-use snafu::OptionExt;
-
-use super::{Error, NoKraftControllersFoundSnafu, kraft_controllers};
+use super::kraft_controllers;
 use crate::{
     crd::{
         KafkaPodDescriptor,
@@ -22,9 +20,8 @@ pub fn build(
     pod_descriptors: &[KafkaPodDescriptor],
     kraft_mode: bool,
     overrides: BTreeMap<String, String>,
-) -> Result<BTreeMap<String, String>, Error> {
-    let kraft_controllers =
-        kraft_controllers(pod_descriptors).context(NoKraftControllersFoundSnafu)?;
+) -> BTreeMap<String, String> {
+    let kraft_controllers = kraft_controllers(pod_descriptors).join(",");
 
     let mut result = BTreeMap::from([
         (
@@ -72,5 +69,5 @@ pub fn build(
     result.extend(graceful_shutdown_config_properties());
     result.extend(overrides);
 
-    Ok(result)
+    result
 }

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

@@ -3,18 +3,10 @@
 pub mod broker_properties;
 pub mod controller_properties;
 
-use snafu::Snafu;
-
 use crate::crd::{KafkaPodDescriptor, role::KafkaRole};
 
-#[derive(Snafu, Debug)]
-pub enum Error {
-    #[snafu(display("no Kraft controllers found to build"))]
-    NoKraftControllersFound,
-}
-
-pub(crate) fn kraft_controllers(pod_descriptors: &[KafkaPodDescriptor]) -> Option<String> {
-    let result = pod_descriptors
+pub(crate) fn kraft_controllers(pod_descriptors: &[KafkaPodDescriptor]) -> Vec<String> {
+    pod_descriptors
         .iter()
         .filter(|pd| pd.role == KafkaRole::Controller.to_string())
         .map(|desc| {
@@ -25,11 +17,4 @@ pub(crate) fn kraft_controllers(pod_descriptors: &[KafkaPodDescriptor]) -> Optio
             )
         })
         .collect::<Vec<String>>()
-        .join(",");
-
-    if result.is_empty() {
-        None
-    } else {
-        Some(result)
-    }
 }

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

@@ -39,6 +39,12 @@ pub enum Error {
 
     #[snafu(display("failed to resolve merged config for rolegroup"))]
     ResolveMergedConfig { source: crate::crd::role::Error },
+
+    #[snafu(display("failed to build pod descriptors"))]
+    BuildPodDescriptors { source: crate::crd::Error },
+
+    #[snafu(display("invalid metadata manager"))]
+    InvalidMetadataManager { source: crate::crd::Error },
 }
 
 type Result<T, E = Error> = std::result::Result<T, E>;
@@ -141,12 +147,25 @@ pub fn validate(
         role_groups.insert(KafkaRole::Controller, controller_groups);
     }
 
+    let pod_descriptors = kafka
+        .pod_descriptors(
+            None,
+            &dereferenced_objects.kubernetes_cluster_info,
+            kafka_security.client_port(),
+        )
+        .context(BuildPodDescriptorsSnafu)?;
+
+    let metadata_manager = kafka
+        .effective_metadata_manager()
+        .context(InvalidMetadataManagerSnafu)?;
+
     Ok(ValidatedKafkaCluster {
         image,
         kafka_security,
         authorization_config: dereferenced_objects.authorization_config,
         role_groups,
-        kubernetes_cluster_info: dereferenced_objects.kubernetes_cluster_info,
+        pod_descriptors,
+        metadata_manager,
     })
 }