refactor: add framework module and merging

Author: maltesander

rust/operator-binary/src/config/jvm.rs

@@ -110,8 +110,17 @@ fn is_heap_jvm_argument(jvm_argument: &str) -> bool {
 
 #[cfg(test)]
 mod tests {
+    use stackable_operator::kube::ResourceExt;
+
     use super::*;
-    use crate::crd::{BrokerRole, role::KafkaRole, v1alpha1};
+    use crate::{
+        crd::{
+            BrokerRole,
+            role::{KafkaRole, broker::BrokerConfig},
+            v1alpha1,
+        },
+        framework::role_utils::with_validated_config,
+    };
 
     #[test]
     fn test_construct_jvm_arguments_defaults() {
@@ -197,12 +206,12 @@ mod tests {
         let kafka: v1alpha1::KafkaCluster =
             serde_yaml::from_str(kafka_cluster).expect("illegal test input");
 
-        let kafka_role = KafkaRole::Broker;
-        let rolegroup_ref = kafka.rolegroup_ref(&kafka_role, "default");
-        let merged_config = kafka_role
-            .merged_config(&kafka, &rolegroup_ref.role_group)
-            .unwrap();
-        let role = kafka.spec.brokers.unwrap();
+        let role = kafka.spec.brokers.clone().unwrap();
+        let role_group = role.role_groups.get("default").unwrap();
+        let default_config =
+            BrokerConfig::default_config(&kafka.name_any(), &KafkaRole::Broker.to_string());
+        let validated = with_validated_config(role_group, &role, &default_config).unwrap();
+        let merged_config = AnyConfig::Broker(validated.config);
 
         (merged_config, role, "default".to_owned())
     }

rust/operator-binary/src/controller.rs

@@ -38,7 +38,7 @@ use crate::{
         self, APP_NAME, KafkaClusterStatus, KafkaPodDescriptor, MetadataManager, OPERATOR_NAME,
         authorization::KafkaAuthorizationConfig,
         listener::get_kafka_listener_config,
-        role::{AnyConfig, KafkaRole},
+        role::{AnyConfig, AnyConfigOverrides, KafkaRole},
         security::KafkaTlsSecurity,
         v1alpha1,
     },
@@ -296,20 +296,18 @@ impl Resource for ValidatedCluster {
     }
 }
 
-pub struct ValidatedRoleGroupConfig {
-    pub merged_config: AnyConfig,
-    // DESIGN DECISION: overrides are resolved into flat maps HERE rather than stored
-    // as the typed KeyValueConfigOverrides and resolved in the per-file builders (the
-    // hdfs-operator pattern). Reason: broker and controller use different override
-    // struct types (KafkaBrokerConfigOverrides vs KafkaControllerConfigOverrides), so a
-    // single typed field would require an enum. Resolving here keeps the build/properties
-    // builders taking plain `BTreeMap<String,String>`. Alternative: an enum over the two
-    // override types threaded to builders that call resolved_overrides() — more types for
-    // no behavioural gain.
-    pub config_file_overrides: BTreeMap<String, String>,
-    pub jvm_security_overrides: BTreeMap<String, String>,
-    pub env_overrides: BTreeMap<String, String>,
-}
+/// 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 (their concrete config and
+/// override types differ). Produced via the local-`framework`
+/// [`with_validated_config`](crate::framework::role_utils::with_validated_config).
+pub type ValidatedRoleGroupConfig = crate::framework::role_utils::RoleGroupConfig<
+    AnyConfig,
+    stackable_operator::role_utils::JavaCommonConfig,
+    AnyConfigOverrides,
+>;
 
 pub async fn reconcile_kafka(
     kafka: Arc<DeserializeGuard<v1alpha1::KafkaCluster>>,
@@ -434,7 +432,7 @@ pub async fn reconcile_kafka(
                 .context(BuildStatefulsetSnafu)?,
             };
 
-            if let AnyConfig::Broker(broker_config) = &validated_rg.merged_config {
+            if let AnyConfig::Broker(broker_config) = &validated_rg.config {
                 let rg_bootstrap_listener = build_broker_rolegroup_bootstrap_listener(
                     kafka,
                     &validated_cluster,

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

@@ -76,8 +76,12 @@ pub fn build_rolegroup_config_map(
 ) -> Result<ConfigMap, Error> {
     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 kafka_config_file_name = validated_rg.config.config_file_name();
+    let config_overrides = validated_rg
+        .config_overrides
+        .config_file_overrides()
+        .overrides
+        .clone();
 
     let opa_connect = validated_cluster
         .cluster_config
@@ -91,7 +95,7 @@ pub fn build_rolegroup_config_map(
         return NoKraftControllersFoundSnafu.fail();
     }
 
-    let kafka_config = match &validated_rg.merged_config {
+    let kafka_config = match &validated_rg.config {
         AnyConfig::Broker(_) => crate::controller::build::properties::broker_properties::build(
             kafka_security,
             listener_config,
@@ -116,7 +120,10 @@ pub fn build_rolegroup_config_map(
         }
     };
 
-    let jvm_sec_props = &validated_rg.jvm_security_overrides;
+    let jvm_sec_props = &validated_rg
+        .config_overrides
+        .security_properties()
+        .overrides;
 
     let mut cm_builder = ConfigMapBuilder::new();
     cm_builder
@@ -179,7 +186,7 @@ pub fn build_rolegroup_config_map(
     let config_data = role_group_config_map_data(
         &resolved_product_image.product_version,
         rolegroup,
-        &validated_rg.merged_config,
+        &validated_rg.config,
     );
     for (file_name, data) in config_data {
         if let Some(data) = data {