refactor: use v2 RoleGroupConfig and optional replicas

Author: maltesander

rust/operator-binary/src/controller.rs

@@ -197,7 +197,9 @@ impl ValidatedCluster {
                     let resource_names = self.resource_names(role, role_group_name);
                     let role_group_statefulset_name = resource_names.stateful_set_name();
                     let role_group_service_name = resource_names.headless_service_name();
-                    for replica in 0..validated_rg.replicas {
+                    // Pods must be predicted from a concrete count (e.g. for KRaft quorum
+                    // voters), so an unset replica count falls back to 1.
+                    for replica in 0..validated_rg.replicas.unwrap_or(1) {
                         pod_descriptors.push(KafkaPodDescriptor {
                             namespace: self.namespace.clone(),
                             role_group_service_name: role_group_service_name.clone(),
@@ -409,27 +411,21 @@ impl Resource for ValidatedCluster {
     }
 }
 
-/// A validated, merged Kafka role-group config.
-///
-/// The merged config fragment is wrapped in [`AnyConfig`] and the merged
-/// `configOverrides` in [`AnyConfigOverrides`], so a single role-agnostic type carries
-/// both broker and controller role groups. Produced from the upstream
-/// [`stackable_operator::v2::role_utils::with_validated_config`] result in
-/// [`validate`](crate::controller::validate). `jvm_argument_overrides` is already merged
-/// (role <- role group) at validation time and applied as-is during build.
+/// The validated, merged per-role-group product config.
 #[derive(Clone, Debug, PartialEq)]
-pub struct ValidatedRoleGroupConfig {
-    pub replicas: u16,
+pub struct ValidatedKafkaConfig {
     pub config: AnyConfig,
-    pub config_overrides: AnyConfigOverrides,
-    pub env_overrides: stackable_operator::v2::builder::pod::container::EnvVarSet,
-    pub pod_overrides: stackable_operator::k8s_openapi::api::core::v1::PodTemplateSpec,
-    pub jvm_argument_overrides:
-        stackable_operator::v2::jvm_argument_overrides::JvmArgumentOverrides,
     /// Validated logging configuration (derived from `config.logging` during validation).
     pub logging: validate::ValidatedLogging,
 }
 
+/// A validated, merged Kafka role-group config.
+pub type ValidatedRoleGroupConfig = stackable_operator::v2::role_utils::RoleGroupConfig<
+    ValidatedKafkaConfig,
+    stackable_operator::v2::role_utils::JavaCommonConfig,
+    AnyConfigOverrides,
+>;
+
 pub struct Ctx {
     pub client: stackable_operator::client::Client,
     pub operator_environment: OperatorEnvironmentOptions,
@@ -621,6 +617,7 @@ pub async fn reconcile_kafka(
             // 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()
@@ -672,7 +669,7 @@ pub async fn reconcile_kafka(
                 .context(BuildStatefulsetSnafu)?,
             };
 
-            if let AnyConfig::Broker(broker_config) = &validated_rg.config {
+            if let AnyConfig::Broker(broker_config) = &validated_rg.config.config {
                 let rg_bootstrap_listener = build_broker_rolegroup_bootstrap_listener(
                     &validated_cluster,
                     kafka_role,

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

@@ -102,7 +102,12 @@ mod tests {
             .get(&KafkaRole::Broker)
             .and_then(|groups| groups.get(&"default".parse().unwrap()))
             .expect("broker default role group should exist");
-        (rg.config.clone(), rg.jvm_argument_overrides.clone())
+        (
+            rg.config.config.clone(),
+            rg.product_specific_common_config
+                .jvm_argument_overrides
+                .clone(),
+        )
     }
 
     #[test]

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

@@ -71,11 +71,11 @@ pub fn build_rolegroup_config_map(
     listener_config: &KafkaListenerConfig,
     vector_config: Option<String>,
 ) -> Result<ConfigMap, Error> {
-    let role = validated_rg.config.kafka_role();
+    let role = validated_rg.config.config.kafka_role();
     let cluster_config = &validated_cluster.cluster_config;
     let kafka_security = &cluster_config.kafka_security;
     let resolved_product_image = &validated_cluster.image;
-    let kafka_config_file_name = config_file_name(&validated_rg.config).to_string();
+    let kafka_config_file_name = config_file_name(&validated_rg.config.config).to_string();
     let config_overrides = validated_rg
         .config_overrides
         .config_file_overrides()
@@ -90,7 +90,7 @@ pub fn build_rolegroup_config_map(
         return NoKraftControllersFoundSnafu.fail();
     }
 
-    let kafka_config = match &validated_rg.config {
+    let kafka_config = match &validated_rg.config.config {
         AnyConfig::Broker(_) => crate::controller::build::properties::broker_properties::build(
             cluster_config,
             listener_config,
@@ -177,7 +177,7 @@ pub fn build_rolegroup_config_map(
 
     let config_data = role_group_config_map_data(
         &resolved_product_image.product_version,
-        &validated_rg.config,
+        &validated_rg.config.config,
     );
     for (file_name, data) in config_data {
         if let Some(data) = data {

rust/operator-binary/src/controller/build/resource/statefulset.rs

@@ -146,7 +146,7 @@ pub fn build_broker_rolegroup_statefulset(
 ) -> Result<StatefulSet, Error> {
     let kafka_security = &validated_cluster.cluster_config.kafka_security;
     let resolved_product_image = &validated_cluster.image;
-    let merged_config = &validated_rg.config;
+    let merged_config = &validated_rg.config.config;
     let resource_names = validated_cluster.resource_names(kafka_role, role_group_name);
     let recommended_labels = validated_cluster.recommended_labels(kafka_role, role_group_name);
     // Used for PVC templates that cannot be modified once they are deployed
@@ -261,7 +261,9 @@ pub fn build_broker_rolegroup_statefulset(
     add_common_kafka_env(
         &mut cb_kafka,
         merged_config,
-        &validated_rg.jvm_argument_overrides,
+        &validated_rg
+            .product_specific_common_config
+            .jvm_argument_overrides,
         resolved_product_image,
         kafka_role,
         role_group_name,
@@ -334,7 +336,11 @@ pub fn build_broker_rolegroup_statefulset(
             ..Probe::default()
         });
 
-    add_log_config_volume(&mut pod_builder, &validated_rg.logging, &resource_names)?;
+    add_log_config_volume(
+        &mut pod_builder,
+        &validated_rg.config.logging,
+        &resource_names,
+    )?;
 
     let metadata = ObjectMetaBuilder::new()
         .with_labels(recommended_labels.clone())
@@ -377,7 +383,7 @@ pub fn build_broker_rolegroup_statefulset(
 
     add_vector_container(
         &mut pod_builder,
-        &validated_rg.logging,
+        &validated_rg.config.logging,
         resolved_product_image,
         &resource_names,
     );
@@ -407,7 +413,7 @@ pub fn build_broker_rolegroup_statefulset(
             .build(),
         spec: Some(StatefulSetSpec {
             pod_management_policy: Some("Parallel".to_string()),
-            replicas: Some(i32::from(validated_rg.replicas)),
+            replicas: validated_rg.replicas.map(i32::from),
             selector: LabelSelector {
                 match_labels: Some(
                     validated_cluster
@@ -435,7 +441,7 @@ pub fn build_controller_rolegroup_statefulset(
 ) -> Result<StatefulSet, Error> {
     let kafka_security = &validated_cluster.cluster_config.kafka_security;
     let resolved_product_image = &validated_cluster.image;
-    let merged_config = &validated_rg.config;
+    let merged_config = &validated_rg.config.config;
     let resource_names = validated_cluster.resource_names(kafka_role, role_group_name);
     let recommended_labels = validated_cluster.recommended_labels(kafka_role, role_group_name);
 
@@ -529,7 +535,9 @@ pub fn build_controller_rolegroup_statefulset(
     add_common_kafka_env(
         &mut cb_kafka,
         merged_config,
-        &validated_rg.jvm_argument_overrides,
+        &validated_rg
+            .product_specific_common_config
+            .jvm_argument_overrides,
         resolved_product_image,
         kafka_role,
         role_group_name,
@@ -569,7 +577,11 @@ pub fn build_controller_rolegroup_statefulset(
             ..Probe::default()
         });
 
-    add_log_config_volume(&mut pod_builder, &validated_rg.logging, &resource_names)?;
+    add_log_config_volume(
+        &mut pod_builder,
+        &validated_rg.config.logging,
+        &resource_names,
+    )?;
 
     let metadata = ObjectMetaBuilder::new()
         .with_labels(recommended_labels.clone())
@@ -600,7 +612,7 @@ pub fn build_controller_rolegroup_statefulset(
 
     add_vector_container(
         &mut pod_builder,
-        &validated_rg.logging,
+        &validated_rg.config.logging,
         resolved_product_image,
         &resource_names,
     );
@@ -630,7 +642,7 @@ pub fn build_controller_rolegroup_statefulset(
                 type_: Some("RollingUpdate".to_string()),
                 ..StatefulSetUpdateStrategy::default()
             }),
-            replicas: Some(i32::from(validated_rg.replicas)),
+            replicas: validated_rg.replicas.map(i32::from),
             selector: LabelSelector {
                 match_labels: Some(
                     validated_cluster

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

@@ -29,8 +29,8 @@ use stackable_operator::{
 
 use crate::{
     controller::{
-        RoleGroupName, ValidatedCluster, ValidatedClusterConfig, ValidatedRoleConfig,
-        ValidatedRoleGroupConfig,
+        RoleGroupName, ValidatedCluster, ValidatedClusterConfig, ValidatedKafkaConfig,
+        ValidatedRoleConfig, ValidatedRoleGroupConfig,
         dereference::DereferencedObjects,
         security::{self, ValidatedKafkaSecurity},
     },
@@ -363,16 +363,20 @@ where
                 validate_logging(&merged.config.config, vector_aggregator_config_map_name)?;
 
             let validated = ValidatedRoleGroupConfig {
-                replicas: merged.replicas.unwrap_or(1),
-                config: wrap_config(merged.config.config),
+                // Passed through as-is (including `None`) so an unset replica count lets a
+                // horizontal autoscaler own the StatefulSet's `.spec.replicas`.
+                replicas: merged.replicas,
+                config: ValidatedKafkaConfig {
+                    config: wrap_config(merged.config.config),
+                    logging,
+                },
                 config_overrides: wrap_overrides(merged.config.config_overrides),
                 env_overrides,
+                // Kafka does not use CLI overrides; the field is carried (and merged upstream)
+                // but unused.
+                cli_overrides: merged.config.cli_overrides,
                 pod_overrides: merged.config.pod_overrides,
-                jvm_argument_overrides: merged
-                    .config
-                    .product_specific_common_config
-                    .jvm_argument_overrides,
-                logging,
+                product_specific_common_config: merged.config.product_specific_common_config,
             };
             let role_group_name = RoleGroupName::from_str(role_group_name).with_context(|_| {
                 ParseRoleGroupNameSnafu {

rust/operator-binary/src/crd/affinity.rs

@@ -64,7 +64,7 @@ mod tests {
             .role_group_configs
             .get(&role)
             .and_then(|groups| groups.get(&"default".parse().unwrap()))
-            .map(|rg| &rg.config)
+            .map(|rg| &rg.config.config)
             .expect("role group should exist");
 
         assert_eq!(