feat: Use separate ConfigOverrides structs for brokers and controllers

Author: siegfriedweber

extra/crds.yaml

@@ -549,16 +549,6 @@ spec:
                           used by `HashMap<String, String>`.
                         nullable: true
                         type: object
-                      controller.properties:
-                        additionalProperties:
-                          type: string
-                        description: |-
-                          Flat key-value overrides for `*.properties`, Hadoop XML, etc.
-
-                          This is backwards-compatible with the existing flat key-value YAML format
-                          used by `HashMap<String, String>`.
-                        nullable: true
-                        type: object
                       security.properties:
                         additionalProperties:
                           type: string
@@ -1176,16 +1166,6 @@ spec:
                                 used by `HashMap<String, String>`.
                               nullable: true
                               type: object
-                            controller.properties:
-                              additionalProperties:
-                                type: string
-                              description: |-
-                                Flat key-value overrides for `*.properties`, Hadoop XML, etc.
-
-                                This is backwards-compatible with the existing flat key-value YAML format
-                                used by `HashMap<String, String>`.
-                              nullable: true
-                              type: object
                             security.properties:
                               additionalProperties:
                                 type: string
@@ -1804,16 +1784,6 @@ spec:
                       and consult the operator specific usage guide documentation for details on the
                       available config files and settings for the specific product.
                     properties:
-                      broker.properties:
-                        additionalProperties:
-                          type: string
-                        description: |-
-                          Flat key-value overrides for `*.properties`, Hadoop XML, etc.
-
-                          This is backwards-compatible with the existing flat key-value YAML format
-                          used by `HashMap<String, String>`.
-                        nullable: true
-                        type: object
                       controller.properties:
                         additionalProperties:
                           type: string
@@ -2263,16 +2233,6 @@ spec:
                             and consult the operator specific usage guide documentation for details on the
                             available config files and settings for the specific product.
                           properties:
-                            broker.properties:
-                              additionalProperties:
-                                type: string
-                              description: |-
-                                Flat key-value overrides for `*.properties`, Hadoop XML, etc.
-
-                                This is backwards-compatible with the existing flat key-value YAML format
-                                used by `HashMap<String, String>`.
-                              nullable: true
-                              type: object
                             controller.properties:
                               additionalProperties:
                                 type: string

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

@@ -60,50 +60,6 @@ pub const STACKABLE_CONFIG_DIR: &str = "/stackable/config";
 pub const STACKABLE_KERBEROS_DIR: &str = "/stackable/kerberos";
 pub const STACKABLE_KERBEROS_KRB5_PATH: &str = "/stackable/kerberos/krb5.conf";
 
-pub type BrokerRole =
-    Role<BrokerConfigFragment, KafkaConfigOverrides, GenericRoleConfig, JavaCommonConfig>;
-pub type ControllerRole =
-    Role<ControllerConfigFragment, KafkaConfigOverrides, GenericRoleConfig, JavaCommonConfig>;
-
-#[derive(Clone, Debug, Default, Deserialize, JsonSchema, PartialEq, Serialize)]
-#[serde(rename_all = "camelCase")]
-pub struct KafkaConfigOverrides {
-    #[serde(
-        default,
-        rename = "broker.properties",
-        skip_serializing_if = "Option::is_none"
-    )]
-    pub broker_properties: Option<KeyValueConfigOverrides>,
-
-    #[serde(
-        default,
-        rename = "controller.properties",
-        skip_serializing_if = "Option::is_none"
-    )]
-    pub controller_properties: Option<KeyValueConfigOverrides>,
-
-    #[serde(
-        default,
-        rename = "security.properties",
-        skip_serializing_if = "Option::is_none"
-    )]
-    pub security_properties: Option<KeyValueConfigOverrides>,
-}
-
-impl KeyValueOverridesProvider for KafkaConfigOverrides {
-    fn get_key_value_overrides(&self, file: &str) -> BTreeMap<String, Option<String>> {
-        let field = match file {
-            role::broker::BROKER_PROPERTIES_FILE => self.broker_properties.as_ref(),
-            role::controller::CONTROLLER_PROPERTIES_FILE => self.controller_properties.as_ref(),
-            JVM_SECURITY_PROPERTIES_FILE => self.security_properties.as_ref(),
-            _ => None,
-        };
-        field
-            .map(KeyValueConfigOverrides::as_product_config_overrides)
-            .unwrap_or_default()
-    }
-}
-
 #[derive(Snafu, Debug)]
 pub enum Error {
     #[snafu(display(
@@ -138,6 +94,20 @@ pub enum Error {
     },
 }
 
+pub type BrokerRole = Role<
+    BrokerConfigFragment,
+    v1alpha1::KafkaBrokerConfigOverrides,
+    GenericRoleConfig,
+    JavaCommonConfig,
+>;
+
+pub type ControllerRole = Role<
+    ControllerConfigFragment,
+    v1alpha1::KafkaControllerConfigOverrides,
+    GenericRoleConfig,
+    JavaCommonConfig,
+>;
+
 #[versioned(
     version(name = "v1alpha1"),
     crates(
@@ -269,6 +239,42 @@ pub mod versioned {
         #[serde(skip_serializing_if = "Option::is_none")]
         pub broker_id_pod_config_map_name: Option<String>,
     }
+
+    #[derive(Clone, Debug, Default, Deserialize, JsonSchema, PartialEq, Serialize)]
+    #[serde(rename_all = "camelCase")]
+    pub struct KafkaBrokerConfigOverrides {
+        #[serde(
+            default,
+            rename = "broker.properties",
+            skip_serializing_if = "Option::is_none"
+        )]
+        pub broker_properties: Option<KeyValueConfigOverrides>,
+
+        #[serde(
+            default,
+            rename = "security.properties",
+            skip_serializing_if = "Option::is_none"
+        )]
+        pub security_properties: Option<KeyValueConfigOverrides>,
+    }
+
+    #[derive(Clone, Debug, Default, Deserialize, JsonSchema, PartialEq, Serialize)]
+    #[serde(rename_all = "camelCase")]
+    pub struct KafkaControllerConfigOverrides {
+        #[serde(
+            default,
+            rename = "controller.properties",
+            skip_serializing_if = "Option::is_none"
+        )]
+        pub controller_properties: Option<KeyValueConfigOverrides>,
+
+        #[serde(
+            default,
+            rename = "security.properties",
+            skip_serializing_if = "Option::is_none"
+        )]
+        pub security_properties: Option<KeyValueConfigOverrides>,
+    }
 }
 
 impl Default for v1alpha1::KafkaClusterConfig {
@@ -285,6 +291,32 @@ impl Default for v1alpha1::KafkaClusterConfig {
     }
 }
 
+impl KeyValueOverridesProvider for v1alpha1::KafkaBrokerConfigOverrides {
+    fn get_key_value_overrides(&self, file: &str) -> BTreeMap<String, Option<String>> {
+        let field = match file {
+            role::broker::BROKER_PROPERTIES_FILE => self.broker_properties.as_ref(),
+            JVM_SECURITY_PROPERTIES_FILE => self.security_properties.as_ref(),
+            _ => None,
+        };
+        field
+            .map(KeyValueConfigOverrides::as_product_config_overrides)
+            .unwrap_or_default()
+    }
+}
+
+impl KeyValueOverridesProvider for v1alpha1::KafkaControllerConfigOverrides {
+    fn get_key_value_overrides(&self, file: &str) -> BTreeMap<String, Option<String>> {
+        let field = match file {
+            role::controller::CONTROLLER_PROPERTIES_FILE => self.controller_properties.as_ref(),
+            JVM_SECURITY_PROPERTIES_FILE => self.security_properties.as_ref(),
+            _ => None,
+        };
+        field
+            .map(KeyValueConfigOverrides::as_product_config_overrides)
+            .unwrap_or_default()
+    }
+}
+
 impl HasStatusCondition for v1alpha1::KafkaCluster {
     fn conditions(&self) -> Vec<ClusterCondition> {
         match &self.status {

rust/operator-binary/src/crd/role/mod.rs

@@ -22,13 +22,10 @@ use strum::{Display, EnumIter, EnumString, IntoEnumIterator};
 
 use crate::{
     config::jvm::{construct_heap_jvm_args, construct_non_heap_jvm_args},
-    crd::{
-        KafkaConfigOverrides,
-        role::{
-            broker::{BROKER_PROPERTIES_FILE, BrokerConfig, BrokerConfigFragment},
-            commons::{CommonConfig, Storage},
-            controller::{CONTROLLER_PROPERTIES_FILE, ControllerConfig},
-        },
+    crd::role::{
+        broker::{BROKER_PROPERTIES_FILE, BrokerConfig, BrokerConfigFragment},
+        commons::{CommonConfig, Storage},
+        controller::{CONTROLLER_PROPERTIES_FILE, ControllerConfig},
     },
     v1alpha1,
 };
@@ -229,16 +226,17 @@ impl KafkaRole {
         rolegroup: &str,
     ) -> Result<String, Error> {
         match self {
-            Self::Broker => {
-                construct_non_heap_jvm_args::<BrokerConfigFragment, KafkaConfigOverrides>(
-                    merged_config,
-                    kafka.broker_role().with_context(|_| MissingRoleSnafu {
-                        role: self.to_string(),
-                    })?,
-                    rolegroup,
-                )
-                .context(ConstructJvmArgumentsSnafu)
-            }
+            Self::Broker => construct_non_heap_jvm_args::<
+                BrokerConfigFragment,
+                v1alpha1::KafkaBrokerConfigOverrides,
+            >(
+                merged_config,
+                kafka.broker_role().with_context(|_| MissingRoleSnafu {
+                    role: self.to_string(),
+                })?,
+                rolegroup,
+            )
+            .context(ConstructJvmArgumentsSnafu),
             Self::Controller => construct_non_heap_jvm_args(
                 merged_config,
                 kafka.controller_role().with_context(|_| MissingRoleSnafu {
@@ -257,7 +255,10 @@ impl KafkaRole {
         rolegroup: &str,
     ) -> Result<String, Error> {
         match self {
-            Self::Broker => construct_heap_jvm_args::<BrokerConfigFragment, KafkaConfigOverrides>(
+            Self::Broker => construct_heap_jvm_args::<
+                BrokerConfigFragment,
+                v1alpha1::KafkaBrokerConfigOverrides,
+            >(
                 merged_config,
                 kafka.broker_role().with_context(|_| MissingRoleSnafu {
                     role: self.to_string(),

rust/operator-binary/src/kafka_controller.rs

@@ -546,9 +546,9 @@ fn validated_product_config(
     product_version: &str,
     product_config: &ProductConfigManager,
 ) -> Result<ValidatedRoleConfigByPropertyKind, Error> {
-    let mut roles = HashMap::new();
+    let mut role_config = HashMap::new();
 
-    roles.insert(
+    let broker_role = [(
         KafkaRole::Broker.to_string(),
         (
             vec![
@@ -564,11 +564,17 @@ fn validated_product_config(
                 })?
                 .erase(),
         ),
-    );
+    )]
+    .into();
+
+    let broker_role_config =
+        transform_all_roles_to_config(kafka, &broker_role).context(GenerateProductConfigSnafu)?;
+
+    role_config.extend(broker_role_config);
 
     // TODO: need this if because controller_role() raises an error
     if kafka.spec.controllers.is_some() {
-        roles.insert(
+        let controller_role = [(
             KafkaRole::Controller.to_string(),
             (
                 vec![
@@ -584,11 +590,14 @@ fn validated_product_config(
                     })?
                     .erase(),
             ),
-        );
-    }
+        )]
+        .into();
+
+        let controller_role_config = transform_all_roles_to_config(kafka, &controller_role)
+            .context(GenerateProductConfigSnafu)?;
 
-    let role_config =
-        transform_all_roles_to_config(kafka, &roles).context(GenerateProductConfigSnafu)?;
+        role_config.extend(controller_role_config);
+    }
 
     validate_all_roles_and_groups_config(
         product_version,