From 3fceb1a5fb4a71799ba4057e764211f2e32ccb9b Mon Sep 17 00:00:00 2001 From: Malte Sander Date: Fri, 10 Jul 2026 15:57:54 +0200 Subject: [PATCH 1/4] refactor: decouple statefulset builders from applied ServiceAccount --- rust/operator-binary/src/controller.rs | 9 ++++++--- .../src/controller/build/resource/statefulset.rs | 15 +++++++-------- 2 files changed, 13 insertions(+), 11 deletions(-) diff --git a/rust/operator-binary/src/controller.rs b/rust/operator-binary/src/controller.rs index 395c1766..2bbc8a4a 100644 --- a/rust/operator-binary/src/controller.rs +++ b/rust/operator-binary/src/controller.rs @@ -20,7 +20,7 @@ use stackable_operator::{ commons::{networking::DomainName, product_image_selection::ResolvedProductImage}, crd::listener, kube::{ - Resource, + Resource, ResourceExt, api::{DynamicObject, ObjectMeta}, core::{DeserializeGuard, error_boundary}, runtime::{controller::Action, reflector::ObjectRef}, @@ -557,6 +557,9 @@ pub async fn reconcile_kafka( .add(client, rbac_sa.clone()) .await .context(ApplyServiceAccountSnafu)?; + // The ServiceAccount name is deterministic, so the statefulset builders only need the name, + // not the applied object. + let service_account_name = rbac_sa.name_any(); cluster_resources .add(client, rbac_rolebinding) .await @@ -608,7 +611,7 @@ pub async fn reconcile_kafka( rolegroup_name, &validated_cluster, validated_rg, - &rbac_sa, + &service_account_name, ) .context(BuildStatefulsetSnafu)?, KafkaRole::Controller => build_controller_rolegroup_statefulset( @@ -616,7 +619,7 @@ pub async fn reconcile_kafka( rolegroup_name, &validated_cluster, validated_rg, - &rbac_sa, + &service_account_name, ) .context(BuildStatefulsetSnafu)?, }; diff --git a/rust/operator-binary/src/controller/build/resource/statefulset.rs b/rust/operator-binary/src/controller/build/resource/statefulset.rs index 8a9abea1..70997d08 100644 --- a/rust/operator-binary/src/controller/build/resource/statefulset.rs +++ b/rust/operator-binary/src/controller/build/resource/statefulset.rs @@ -20,12 +20,11 @@ use stackable_operator::{ apps::v1::{StatefulSet, StatefulSetSpec, StatefulSetUpdateStrategy}, core::v1::{ ConfigMapVolumeSource, ContainerPort, EnvVar, EnvVarSource, ExecAction, - ObjectFieldSelector, PodSpec, Probe, ServiceAccount, TCPSocketAction, Volume, + ObjectFieldSelector, PodSpec, Probe, TCPSocketAction, Volume, }, }, apimachinery::pkg::{apis::meta::v1::LabelSelector, util::intstr::IntOrString}, }, - kube::ResourceExt, product_logging, v2::{ builder::{ @@ -189,7 +188,7 @@ pub fn build_broker_rolegroup_statefulset( role_group_name: &RoleGroupName, validated_cluster: &ValidatedCluster, validated_rg: &ValidatedRoleGroupConfig, - service_account: &ServiceAccount, + service_account_name: &str, ) -> Result { let kafka_security = &validated_cluster.cluster_config.kafka_security; let resolved_product_image = &validated_cluster.image; @@ -395,7 +394,7 @@ pub fn build_broker_rolegroup_statefulset( .add_container(cb_kcat_prober.build()) .affinity(&merged_config.affinity); - add_common_pod_config(&mut pod_builder, &resource_names, service_account)?; + add_common_pod_config(&mut pod_builder, &resource_names, service_account_name)?; add_vector_container( &mut pod_builder, @@ -453,7 +452,7 @@ pub fn build_controller_rolegroup_statefulset( role_group_name: &RoleGroupName, validated_cluster: &ValidatedCluster, validated_rg: &ValidatedRoleGroupConfig, - service_account: &ServiceAccount, + service_account_name: &str, ) -> Result { let kafka_security = &validated_cluster.cluster_config.kafka_security; let resolved_product_image = &validated_cluster.image; @@ -578,7 +577,7 @@ pub fn build_controller_rolegroup_statefulset( .add_container(kafka_container) .affinity(&merged_config.affinity); - add_common_pod_config(&mut pod_builder, &resource_names, service_account)?; + add_common_pod_config(&mut pod_builder, &resource_names, service_account_name)?; add_vector_container( &mut pod_builder, @@ -729,7 +728,7 @@ fn add_log_config_volume( fn add_common_pod_config( pod_builder: &mut PodBuilder, resource_names: &ResourceNames, - service_account: &ServiceAccount, + service_account_name: &str, ) -> Result<(), Error> { pod_builder .add_volume(Volume { @@ -748,7 +747,7 @@ fn add_common_pod_config( )), ) .context(AddVolumeSnafu)? - .service_account_name(service_account.name_any()) + .service_account_name(service_account_name) .security_context(PodSecurityContextBuilder::new().fs_group(1000).build()); Ok(()) } From 3d321ba08a98f209367f73c6b974eb66aaf5df7f Mon Sep 17 00:00:00 2001 From: Malte Sander Date: Fri, 10 Jul 2026 16:03:38 +0200 Subject: [PATCH 2/4] refactor: read cluster_domain from ValidatedCluster in listener config --- rust/operator-binary/src/controller.rs | 1 - .../controller/build/properties/listener.rs | 27 +++++++------------ 2 files changed, 9 insertions(+), 19 deletions(-) diff --git a/rust/operator-binary/src/controller.rs b/rust/operator-binary/src/controller.rs index 2bbc8a4a..46f3aadc 100644 --- a/rust/operator-binary/src/controller.rs +++ b/rust/operator-binary/src/controller.rs @@ -593,7 +593,6 @@ pub async fn reconcile_kafka( &validated_cluster.cluster_config.kafka_security, kafka_role, rolegroup_name, - &client.kubernetes_cluster_info, ); let rg_configmap = build::resource::config_map::build_rolegroup_config_map( diff --git a/rust/operator-binary/src/controller/build/properties/listener.rs b/rust/operator-binary/src/controller/build/properties/listener.rs index 9b5953fe..d58cba74 100644 --- a/rust/operator-binary/src/controller/build/properties/listener.rs +++ b/rust/operator-binary/src/controller/build/properties/listener.rs @@ -6,9 +6,7 @@ use std::collections::BTreeMap; -use stackable_operator::{ - utils::cluster_info::KubernetesClusterInfo, v2::types::kubernetes::NamespaceName, -}; +use stackable_operator::{commons::networking::DomainName, v2::types::kubernetes::NamespaceName}; use crate::{ controller::{RoleGroupName, ValidatedCluster, security::ValidatedKafkaSecurity}, @@ -27,7 +25,6 @@ pub fn get_kafka_listener_config( kafka_security: &ValidatedKafkaSecurity, role: &KafkaRole, role_group_name: &RoleGroupName, - cluster_info: &KubernetesClusterInfo, ) -> KafkaListenerConfig { let headless_service_name = validated_cluster .resource_names(role, role_group_name) @@ -35,7 +32,7 @@ pub fn get_kafka_listener_config( let pod_fqdn = pod_fqdn( &validated_cluster.namespace, headless_service_name.as_ref(), - cluster_info, + &validated_cluster.cluster_domain, ); let mut listeners = vec![]; let mut advertised_listeners = vec![]; @@ -143,12 +140,9 @@ pub fn get_kafka_listener_config( pub(crate) fn pod_fqdn( namespace: &NamespaceName, sts_service_name: &str, - cluster_info: &KubernetesClusterInfo, + cluster_domain: &DomainName, ) -> String { - format!( - "${{env:POD_NAME}}.{sts_service_name}.{namespace}.svc.{cluster_domain}", - cluster_domain = cluster_info.cluster_domain - ) + format!("${{env:POD_NAME}}.{sts_service_name}.{namespace}.svc.{cluster_domain}") } #[cfg(test)] @@ -157,6 +151,7 @@ mod tests { builder::meta::ObjectMetaBuilder, commons::networking::DomainName, crd::authentication::{core, kerberos, tls}, + utils::cluster_info::KubernetesClusterInfo, }; use super::*; @@ -217,7 +212,6 @@ mod tests { &kafka_security, &KafkaRole::Broker, &role_group_name, - &cluster_info, ); assert_eq!( @@ -250,7 +244,7 @@ mod tests { .resource_names(&KafkaRole::Broker, &role_group_name) .headless_service_name() .as_ref(), - &cluster_info + &cluster_info.cluster_domain ), internal_port = kafka_security.internal_port(), ) @@ -280,7 +274,6 @@ mod tests { &kafka_security, &KafkaRole::Broker, &role_group_name, - &cluster_info, ); assert_eq!( @@ -313,7 +306,7 @@ mod tests { .resource_names(&KafkaRole::Broker, &role_group_name) .headless_service_name() .as_ref(), - &cluster_info + &cluster_info.cluster_domain ), internal_port = kafka_security.internal_port(), ) @@ -344,7 +337,6 @@ mod tests { &kafka_security, &KafkaRole::Broker, &role_group_name, - &cluster_info, ); assert_eq!( @@ -377,7 +369,7 @@ mod tests { .resource_names(&KafkaRole::Broker, &role_group_name) .headless_service_name() .as_ref(), - &cluster_info + &cluster_info.cluster_domain ), internal_port = kafka_security.internal_port(), ) @@ -442,7 +434,6 @@ mod tests { &kafka_security, &KafkaRole::Broker, &role_group_name, - &cluster_info, ); assert_eq!( @@ -478,7 +469,7 @@ mod tests { .resource_names(&KafkaRole::Broker, &role_group_name) .headless_service_name() .as_ref(), - &cluster_info + &cluster_info.cluster_domain ), internal_port = kafka_security.internal_port(), bootstrap_name = KafkaListenerName::Bootstrap, From 6c06c70fd15df1046573d4cf5ea5ed83c9ef6c1d Mon Sep 17 00:00:00 2001 From: Malte Sander Date: Fri, 10 Jul 2026 16:12:23 +0200 Subject: [PATCH 3/4] refactor: introduce build() aggregator for Kubernetes resources --- rust/operator-binary/src/controller.rs | 230 ++++++------------ .../src/controller/build/mod.rs | 150 ++++++++++++ 2 files changed, 220 insertions(+), 160 deletions(-) diff --git a/rust/operator-binary/src/controller.rs b/rust/operator-binary/src/controller.rs index 46f3aadc..0371fb63 100644 --- a/rust/operator-binary/src/controller.rs +++ b/rust/operator-binary/src/controller.rs @@ -19,6 +19,11 @@ use stackable_operator::{ cluster_resources::ClusterResourceApplyStrategy, commons::{networking::DomainName, product_image_selection::ResolvedProductImage}, crd::listener, + k8s_openapi::api::{ + apps::v1::StatefulSet, + core::v1::{ConfigMap, Service}, + policy::v1::PodDisruptionBudget, + }, kube::{ Resource, ResourceExt, api::{DynamicObject, ObjectMeta}, @@ -55,18 +60,7 @@ pub use stackable_operator::v2::types::operator::{RoleGroupName, RoleName}; use crate::{ controller::{ - build::{ - properties::listener::get_kafka_listener_config, - resource::{ - listener::build_broker_rolegroup_bootstrap_listener, - pdb::build_pdb, - rbac::{build_rbac_role_binding, build_rbac_service_account}, - service::{build_rolegroup_headless_service, build_rolegroup_metrics_service}, - statefulset::{ - build_broker_rolegroup_statefulset, build_controller_rolegroup_statefulset, - }, - }, - }, + build::resource::rbac::{build_rbac_role_binding, build_rbac_service_account}, node_id_hasher::node_id_hash32_offset, security::ValidatedKafkaSecurity, }, @@ -94,6 +88,19 @@ pub enum PodDescriptorsError { }, } +/// Every Kubernetes resource produced by the [`build`] step. +/// +/// The discovery `ConfigMap` is not part of this: it depends on the applied bootstrap +/// [`Listener`](listener)s' status and is therefore built in [`reconcile_kafka`] after they are +/// applied. +pub struct KubernetesResources { + pub stateful_sets: Vec, + pub services: Vec, + pub listeners: Vec, + pub config_maps: Vec, + pub pod_disruption_budgets: Vec, +} + /// The validated cluster. Carries everything the build steps need, resolved once /// here so downstream code never re-derives it or touches the raw spec. /// @@ -393,27 +400,12 @@ pub enum Error { #[snafu(display("failed to validate cluster"))] ValidateCluster { source: validate::Error }, - #[snafu(display("failed to apply bootstrap Listener"))] - ApplyBootstrapListener { - source: stackable_operator::cluster_resources::Error, - }, + #[snafu(display("failed to build the Kubernetes resources"))] + BuildResources { source: build::Error }, - #[snafu(display("failed to apply Service for role group {role_group}"))] - ApplyRoleGroupService { + #[snafu(display("failed to apply Kubernetes resource"))] + ApplyResource { source: stackable_operator::cluster_resources::Error, - role_group: RoleGroupName, - }, - - #[snafu(display("failed to apply ConfigMap for role group {role_group}"))] - ApplyRoleGroupConfig { - source: stackable_operator::cluster_resources::Error, - role_group: RoleGroupName, - }, - - #[snafu(display("failed to apply StatefulSet for role group {role_group}"))] - ApplyRoleGroupStatefulSet { - source: stackable_operator::cluster_resources::Error, - role_group: RoleGroupName, }, #[snafu(display("failed to build discovery ConfigMap"))] @@ -446,11 +438,6 @@ pub enum Error { source: stackable_operator::client::Error, }, - #[snafu(display("failed to apply PodDisruptionBudget"))] - ApplyPdb { - source: stackable_operator::cluster_resources::Error, - }, - #[snafu(display("failed to get required Labels"))] GetRequiredLabels { source: @@ -461,16 +448,6 @@ pub enum Error { InvalidKafkaCluster { source: error_boundary::InvalidObject, }, - - #[snafu(display("failed to build statefulset"))] - BuildStatefulset { - source: crate::controller::build::resource::statefulset::Error, - }, - - #[snafu(display("failed to build configmap"))] - BuildConfigMap { - source: crate::controller::build::resource::config_map::Error, - }, } type Result = std::result::Result; @@ -483,21 +460,16 @@ impl ReconcilerError for Error { match self { Error::Dereference { .. } => None, Error::ValidateCluster { .. } => None, - Error::ApplyBootstrapListener { .. } => None, - Error::ApplyRoleGroupService { .. } => None, - Error::ApplyRoleGroupConfig { .. } => None, - Error::ApplyRoleGroupStatefulSet { .. } => None, + Error::BuildResources { .. } => None, + Error::ApplyResource { .. } => None, Error::BuildDiscoveryConfig { .. } => None, Error::ApplyDiscoveryConfig { .. } => None, Error::DeleteOrphans { .. } => None, Error::ApplyServiceAccount { .. } => None, Error::ApplyRoleBinding { .. } => None, Error::ApplyStatus { .. } => None, - Error::ApplyPdb { .. } => None, Error::GetRequiredLabels { .. } => None, Error::InvalidKafkaCluster { .. } => None, - Error::BuildStatefulset { .. } => None, - Error::BuildConfigMap { .. } => None, } } } @@ -565,121 +537,59 @@ pub async fn reconcile_kafka( .await .context(ApplyRoleBindingSnafu)?; - let mut bootstrap_listeners = Vec::::new(); - - for (kafka_role, rg_map) in &validated_cluster.role_group_configs { - for (rolegroup_name, validated_rg) in rg_map { - // The Vector agent config is the static `vector.yaml`, added to the rolegroup - // ConfigMap only when the Vector agent is enabled (resolved during validation). - let vector_config = validated_rg - .config - .logging - .vector_container - .is_some() - .then(build::properties::product_logging::vector_config_file_content); - - let rg_headless_service = build_rolegroup_headless_service( - &validated_cluster, - kafka_role, - rolegroup_name, - &validated_cluster.cluster_config.kafka_security, - ); - - let rg_metrics_service = - build_rolegroup_metrics_service(&validated_cluster, kafka_role, rolegroup_name); - - let kafka_listeners = get_kafka_listener_config( - &validated_cluster, - &validated_cluster.cluster_config.kafka_security, - kafka_role, - rolegroup_name, - ); - - let rg_configmap = build::resource::config_map::build_rolegroup_config_map( - &validated_cluster, - rolegroup_name, - validated_rg, - &kafka_listeners, - vector_config, - ) - .context(BuildConfigMapSnafu)?; - - let rg_statefulset = match kafka_role { - KafkaRole::Broker => build_broker_rolegroup_statefulset( - kafka_role, - rolegroup_name, - &validated_cluster, - validated_rg, - &service_account_name, - ) - .context(BuildStatefulsetSnafu)?, - KafkaRole::Controller => build_controller_rolegroup_statefulset( - kafka_role, - rolegroup_name, - &validated_cluster, - validated_rg, - &service_account_name, - ) - .context(BuildStatefulsetSnafu)?, - }; - - if let AnyConfig::Broker(broker_config) = &validated_rg.config.config { - let rg_bootstrap_listener = build_broker_rolegroup_bootstrap_listener( - &validated_cluster, - kafka_role, - rolegroup_name, - broker_config, - ); - bootstrap_listeners.push( - cluster_resources - .add(client, rg_bootstrap_listener) - .await - .context(ApplyBootstrapListenerSnafu)?, - ); - } + // Build every Kubernetes resource up front (client-free). The discovery ConfigMap is not part + // of this, as it depends on the applied bootstrap Listeners' status (see below). + let resources = + build::build(&validated_cluster, &service_account_name).context(BuildResourcesSnafu)?; + + // Apply order: Services, then Listeners (collecting the applied bootstrap Listeners for the + // discovery ConfigMap), then ConfigMaps, then PodDisruptionBudgets, and finally the + // StatefulSets. The StatefulSets must be applied after all ConfigMaps and Secrets they mount to + // prevent unnecessary Pod restarts. + // See https://github.com/stackabletech/commons-operator/issues/111 for details. + for service in resources.services { + cluster_resources + .add(client, service) + .await + .context(ApplyResourceSnafu)?; + } + let mut bootstrap_listeners = Vec::::new(); + for rg_listener in resources.listeners { + bootstrap_listeners.push( cluster_resources - .add(client, rg_headless_service) - .await - .with_context(|_| ApplyRoleGroupServiceSnafu { - role_group: rolegroup_name.clone(), - })?; - cluster_resources - .add(client, rg_metrics_service) - .await - .with_context(|_| ApplyRoleGroupServiceSnafu { - role_group: rolegroup_name.clone(), - })?; - cluster_resources - .add(client, rg_configmap) + .add(client, rg_listener) .await - .with_context(|_| ApplyRoleGroupConfigSnafu { - role_group: rolegroup_name.clone(), - })?; - - // Note: The StatefulSet needs to be applied after all ConfigMaps and Secrets it mounts - // to prevent unnecessary Pod restarts. - // See https://github.com/stackabletech/commons-operator/issues/111 for details. - ss_cond_builder.add( - cluster_resources - .add(client, rg_statefulset) - .await - .with_context(|_| ApplyRoleGroupStatefulSetSnafu { - role_group: rolegroup_name.clone(), - })?, - ); - } + .context(ApplyResourceSnafu)?, + ); + } + + for config_map in resources.config_maps { + cluster_resources + .add(client, config_map) + .await + .context(ApplyResourceSnafu)?; + } + + for pdb in resources.pod_disruption_budgets { + cluster_resources + .add(client, pdb) + .await + .context(ApplyResourceSnafu)?; + } - if let Some(role_config) = validated_cluster.role_configs.get(kafka_role) - && let Some(pdb) = build_pdb(&role_config.pdb, &validated_cluster, kafka_role) - { + for stateful_set in resources.stateful_sets { + ss_cond_builder.add( cluster_resources - .add(client, pdb) + .add(client, stateful_set) .await - .context(ApplyPdbSnafu)?; - } + .context(ApplyResourceSnafu)?, + ); } + // The discovery ConfigMap reports the bootstrap Listeners' ingress addresses, which are only + // populated on the applied Listener objects (by the Listener operator), so it is built here + // rather than in the client-free build() step. let discovery_cm = build::resource::discovery::build_discovery_configmap( &validated_cluster, &bootstrap_listeners, diff --git a/rust/operator-binary/src/controller/build/mod.rs b/rust/operator-binary/src/controller/build/mod.rs index fc57c343..05cfa4b3 100644 --- a/rust/operator-binary/src/controller/build/mod.rs +++ b/rust/operator-binary/src/controller/build/mod.rs @@ -1,5 +1,28 @@ //! Builders that assemble Kubernetes resources for kafka rolegroups. +use snafu::{ResultExt, Snafu}; + +use crate::{ + controller::{ + KubernetesResources, RoleGroupName, ValidatedCluster, + build::{ + properties::{ + listener::get_kafka_listener_config, product_logging::vector_config_file_content, + }, + resource::{ + config_map::build_rolegroup_config_map, + listener::build_broker_rolegroup_bootstrap_listener, + pdb::build_pdb, + service::{build_rolegroup_headless_service, build_rolegroup_metrics_service}, + statefulset::{ + build_broker_rolegroup_statefulset, build_controller_rolegroup_statefulset, + }, + }, + }, + }, + crd::role::{AnyConfig, KafkaRole}, +}; + pub mod command; pub mod graceful_shutdown; pub mod jvm; @@ -8,3 +31,130 @@ pub mod labels; pub mod properties; pub mod resource; pub mod security; + +#[derive(Snafu, Debug)] +pub enum Error { + #[snafu(display("failed to build ConfigMap for role group {role_group}"))] + ConfigMap { + source: resource::config_map::Error, + role_group: RoleGroupName, + }, + + #[snafu(display("failed to build StatefulSet for role group {role_group}"))] + StatefulSet { + source: resource::statefulset::Error, + role_group: RoleGroupName, + }, +} + +/// Builds every Kubernetes resource for the given validated cluster. +/// +/// Does not need a Kubernetes client: every external reference is already dereferenced and +/// validated by this point, so the only errors are resource-assembly failures. +/// +/// The discovery `ConfigMap` is intentionally excluded: it reports the applied bootstrap +/// `Listener`s' ingress addresses (populated by the Listener operator only after apply), so it is +/// built in the reconcile step once those `Listener`s exist. +/// +/// `service_account_name` is the name of the RBAC `ServiceAccount` the role-group Pods run under. +/// The RBAC resources are built and applied separately, in the reconcile step; the name is +/// deterministic, so the build step does not depend on the applied `ServiceAccount`. +pub fn build( + cluster: &ValidatedCluster, + service_account_name: &str, +) -> Result { + let mut stateful_sets = vec![]; + let mut services = vec![]; + let mut listeners = vec![]; + let mut config_maps = vec![]; + let mut pod_disruption_budgets = vec![]; + + for (role, role_group_configs) in &cluster.role_group_configs { + // Kafka's `GenericRoleConfig` only carries the PodDisruptionBudget. + if let Some(role_config) = cluster.role_configs.get(role) { + pod_disruption_budgets.extend(build_pdb(&role_config.pdb, cluster, role)); + } + + for (role_group_name, validated_rg) in role_group_configs { + // The Vector agent config is the static `vector.yaml`, added to the rolegroup + // ConfigMap only when the Vector agent is enabled (resolved during validation). + let vector_config = validated_rg + .config + .logging + .vector_container + .is_some() + .then(vector_config_file_content); + + services.push(build_rolegroup_headless_service( + cluster, + role, + role_group_name, + &cluster.cluster_config.kafka_security, + )); + services.push(build_rolegroup_metrics_service( + cluster, + role, + role_group_name, + )); + + let kafka_listeners = get_kafka_listener_config( + cluster, + &cluster.cluster_config.kafka_security, + role, + role_group_name, + ); + + config_maps.push( + build_rolegroup_config_map( + cluster, + role_group_name, + validated_rg, + &kafka_listeners, + vector_config, + ) + .context(ConfigMapSnafu { + role_group: role_group_name.clone(), + })?, + ); + + let stateful_set = match role { + KafkaRole::Broker => build_broker_rolegroup_statefulset( + role, + role_group_name, + cluster, + validated_rg, + service_account_name, + ), + KafkaRole::Controller => build_controller_rolegroup_statefulset( + role, + role_group_name, + cluster, + validated_rg, + service_account_name, + ), + } + .context(StatefulSetSnafu { + role_group: role_group_name.clone(), + })?; + stateful_sets.push(stateful_set); + + // Only broker role groups get a bootstrap Listener. + if let AnyConfig::Broker(broker_config) = &validated_rg.config.config { + listeners.push(build_broker_rolegroup_bootstrap_listener( + cluster, + role, + role_group_name, + broker_config, + )); + } + } + } + + Ok(KubernetesResources { + stateful_sets, + services, + listeners, + config_maps, + pod_disruption_budgets, + }) +} From e7124097cca43faf4f276282aaa9b8130a86d917 Mon Sep 17 00:00:00 2001 From: Malte Sander Date: Fri, 10 Jul 2026 16:17:48 +0200 Subject: [PATCH 4/4] test: cover the build() aggregator's resource output --- .../src/controller/build/mod.rs | 146 ++++++++++++++++++ 1 file changed, 146 insertions(+) diff --git a/rust/operator-binary/src/controller/build/mod.rs b/rust/operator-binary/src/controller/build/mod.rs index 05cfa4b3..f33b2f30 100644 --- a/rust/operator-binary/src/controller/build/mod.rs +++ b/rust/operator-binary/src/controller/build/mod.rs @@ -158,3 +158,149 @@ pub fn build( pod_disruption_budgets, }) } + +#[cfg(test)] +mod tests { + use stackable_operator::kube::Resource; + + use super::build; + use crate::controller::{ + ValidatedCluster, + test_support::{minimal_kafka, validated_cluster}, + }; + + /// Sorted `metadata.name`s of the given resources, for order-independent assertions. + fn sorted_names(resources: &[impl Resource]) -> Vec<&str> { + let mut names: Vec<&str> = resources + .iter() + .filter_map(|resource| resource.meta().name.as_deref()) + .collect(); + names.sort(); + names + } + + /// A KRaft cluster with one `broker` and one `controller` role group, resolved through the real + /// validate step (mirroring the other build fixtures), since [`ValidatedCluster`] carries + /// several resolved types that are impractical to construct by hand. + fn kraft_cluster() -> ValidatedCluster { + let kafka = minimal_kafka( + r#" + apiVersion: kafka.stackable.tech/v1alpha1 + kind: KafkaCluster + metadata: + name: simple-kafka + namespace: default + uid: 12345678-1234-1234-1234-123456789012 + spec: + image: + productVersion: 3.9.2 + clusterConfig: + metadataManager: kraft + controllers: + roleGroups: + default: + replicas: 3 + brokers: + roleGroups: + default: + replicas: 3 + "#, + ); + validated_cluster(&kafka) + } + + /// A ZooKeeper-mode cluster with a single `broker` role group (no controllers). + fn zookeeper_cluster() -> ValidatedCluster { + let kafka = minimal_kafka( + r#" + apiVersion: kafka.stackable.tech/v1alpha1 + kind: KafkaCluster + metadata: + name: simple-kafka + namespace: default + uid: 12345678-1234-1234-1234-123456789012 + spec: + image: + productVersion: 3.9.2 + clusterConfig: + zookeeperConfigMapName: xyz + brokers: + roleGroups: + default: + replicas: 1 + "#, + ); + validated_cluster(&kafka) + } + + #[test] + fn build_produces_expected_resource_names() { + let cluster = kraft_cluster(); + let resources = build(&cluster, "simple-kafka-serviceaccount").expect("build succeeds"); + + // One StatefulSet per role group. + assert_eq!( + sorted_names(&resources.stateful_sets), + [ + "simple-kafka-broker-default", + "simple-kafka-controller-default" + ] + ); + // One rolegroup ConfigMap per role group. + assert_eq!( + sorted_names(&resources.config_maps), + [ + "simple-kafka-broker-default", + "simple-kafka-controller-default" + ] + ); + // One headless and one metrics Service per role group. + assert_eq!( + sorted_names(&resources.services), + [ + "simple-kafka-broker-default-headless", + "simple-kafka-broker-default-metrics", + "simple-kafka-controller-default-headless", + "simple-kafka-controller-default-metrics", + ] + ); + // Only broker role groups get a bootstrap Listener. + assert_eq!( + sorted_names(&resources.listeners), + ["simple-kafka-broker-default-bootstrap"] + ); + // A default PodDisruptionBudget per role. + assert_eq!( + sorted_names(&resources.pod_disruption_budgets), + ["simple-kafka-broker", "simple-kafka-controller"] + ); + } + + /// ZooKeeper mode has no `controller` role, so `build()` emits no controller resources while + /// still producing the broker's bootstrap Listener. + #[test] + fn build_zookeeper_mode_has_no_controller_resources() { + let cluster = zookeeper_cluster(); + let resources = build(&cluster, "simple-kafka-serviceaccount").expect("build succeeds"); + + assert_eq!( + sorted_names(&resources.stateful_sets), + ["simple-kafka-broker-default"] + ); + assert_eq!( + sorted_names(&resources.services), + [ + "simple-kafka-broker-default-headless", + "simple-kafka-broker-default-metrics", + ] + ); + assert_eq!( + sorted_names(&resources.listeners), + ["simple-kafka-broker-default-bootstrap"] + ); + assert_eq!( + sorted_names(&resources.pod_disruption_budgets), + ["simple-kafka-broker"] + ); + } +}