refactor: make better use of ValidatedCluster in property file builders

Author: maltesander

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

@@ -13,8 +13,7 @@ use crate::{
         build::properties::logging::role_group_config_map_data,
     },
     crd::{
-        ConfigFileName, MetadataManager, STACKABLE_LISTENER_BOOTSTRAP_DIR,
-        STACKABLE_LISTENER_BROKER_DIR,
+        ConfigFileName, STACKABLE_LISTENER_BOOTSTRAP_DIR, STACKABLE_LISTENER_BROKER_DIR,
         listener::{KafkaListenerConfig, node_address_cmd},
         role::AnyConfig,
         v1alpha1,
@@ -63,7 +62,8 @@ pub fn build_rolegroup_config_map(
     validated_rg: &ValidatedRoleGroupConfig,
     listener_config: &KafkaListenerConfig,
 ) -> Result<ConfigMap, Error> {
-    let kafka_security = &validated_cluster.cluster_config.kafka_security;
+    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 = validated_rg.config.config_file_name().to_string();
     let config_overrides = validated_rg
@@ -72,36 +72,20 @@ pub fn build_rolegroup_config_map(
         .overrides
         .clone();
 
-    let opa_connect = validated_cluster
-        .cluster_config
-        .authorization_config
-        .as_ref()
-        .map(|auth_config| auth_config.opa_connect.clone());
-
-    let kraft_mode = validated_cluster.cluster_config.metadata_manager == MetadataManager::KRaft;
-
-    if kraft_mode && validated_cluster.cluster_config.pod_descriptors.is_empty() {
+    if cluster_config.is_kraft_mode() && cluster_config.pod_descriptors.is_empty() {
         return NoKraftControllersFoundSnafu.fail();
     }
 
     let kafka_config = match &validated_rg.config {
         AnyConfig::Broker(_) => crate::controller::build::properties::broker_properties::build(
-            kafka_security,
+            cluster_config,
             listener_config,
-            &validated_cluster.cluster_config.pod_descriptors,
-            opa_connect.as_deref(),
-            kraft_mode,
-            validated_cluster
-                .cluster_config
-                .disable_broker_id_generation,
             config_overrides,
         ),
         AnyConfig::Controller(_) => {
             crate::controller::build::properties::controller_properties::build(
-                kafka_security,
+                cluster_config,
                 listener_config,
-                &validated_cluster.cluster_config.pod_descriptors,
-                kraft_mode,
                 config_overrides,
             )
         }

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

@@ -2,29 +2,24 @@ use std::collections::BTreeMap;
 
 use super::kraft_controllers;
 use crate::{
+    controller::ValidatedClusterConfig,
     crd::{
-        KafkaPodDescriptor,
         listener::{KafkaListenerConfig, KafkaListenerName},
         role::{
             KAFKA_ADVERTISED_LISTENERS, KAFKA_BROKER_ID, KAFKA_CONTROLLER_QUORUM_BOOTSTRAP_SERVERS,
             KAFKA_LISTENER_SECURITY_PROTOCOL_MAP, KAFKA_LISTENERS, KAFKA_LOG_DIRS, KAFKA_NODE_ID,
             KAFKA_PROCESS_ROLES, KafkaRole,
         },
-        security::KafkaTlsSecurity,
     },
     operations::graceful_shutdown::graceful_shutdown_config_properties,
 };
 
 pub fn build(
-    kafka_security: &KafkaTlsSecurity,
+    cluster_config: &ValidatedClusterConfig,
     listener_config: &KafkaListenerConfig,
-    pod_descriptors: &[KafkaPodDescriptor],
-    opa_connect_string: Option<&str>,
-    kraft_mode: bool,
-    disable_broker_id_generation: bool,
     overrides: BTreeMap<String, String>,
 ) -> BTreeMap<String, String> {
-    let kraft_controllers = kraft_controllers(pod_descriptors);
+    let kraft_controllers = kraft_controllers(&cluster_config.pod_descriptors);
 
     let mut result = BTreeMap::from([
         (
@@ -46,7 +41,7 @@ pub fn build(
         ),
     ]);
 
-    if kraft_mode {
+    if cluster_config.is_kraft_mode() {
         let kraft_controllers = kraft_controllers.join(",");
 
         // Running in KRaft mode
@@ -79,7 +74,7 @@ pub fn build(
         // so we disable automatic id generation.
         // This check ensures that existing clusters running in ZooKeeper mode do not
         // suddenly break after the introduction of this change.
-        if disable_broker_id_generation {
+        if cluster_config.disable_broker_id_generation {
             result.extend([
                 (
                     "broker.id.generation.enable".to_string(),
@@ -91,7 +86,7 @@ pub fn build(
     }
 
     // Enable OPA authorization
-    if opa_connect_string.is_some() {
+    if let Some(opa_connect_string) = cluster_config.opa_connect() {
         result.extend([
             (
                 "authorizer.class.name".to_string(),
@@ -103,12 +98,12 @@ pub fn build(
             ),
             (
                 "opa.authorizer.url".to_string(),
-                opa_connect_string.unwrap_or_default().to_string(),
+                opa_connect_string.to_string(),
             ),
         ]);
     }
 
-    result.extend(kafka_security.broker_config_settings());
+    result.extend(cluster_config.kafka_security.broker_config_settings());
     result.extend(graceful_shutdown_config_properties());
     result.extend(overrides);
 

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

@@ -2,26 +2,23 @@ use std::collections::BTreeMap;
 
 use super::kraft_controllers;
 use crate::{
+    controller::ValidatedClusterConfig,
     crd::{
-        KafkaPodDescriptor,
         listener::{KafkaListenerConfig, KafkaListenerName},
         role::{
             KAFKA_CONTROLLER_QUORUM_BOOTSTRAP_SERVERS, KAFKA_LISTENER_SECURITY_PROTOCOL_MAP,
             KAFKA_LISTENERS, KAFKA_LOG_DIRS, KAFKA_NODE_ID, KAFKA_PROCESS_ROLES, KafkaRole,
         },
-        security::KafkaTlsSecurity,
     },
     operations::graceful_shutdown::graceful_shutdown_config_properties,
 };
 
 pub fn build(
-    kafka_security: &KafkaTlsSecurity,
+    cluster_config: &ValidatedClusterConfig,
     listener_config: &KafkaListenerConfig,
-    pod_descriptors: &[KafkaPodDescriptor],
-    kraft_mode: bool,
     overrides: BTreeMap<String, String>,
 ) -> BTreeMap<String, String> {
-    let kraft_controllers = kraft_controllers(pod_descriptors).join(",");
+    let kraft_controllers = kraft_controllers(&cluster_config.pod_descriptors).join(",");
 
     let mut result = BTreeMap::from([
         (
@@ -58,14 +55,14 @@ pub fn build(
 
     // The ZooKeeper connection is needed for migration from ZooKeeper to KRaft mode.
     // It is not needed once the controller is fully running in KRaft mode.
-    if !kraft_mode {
+    if !cluster_config.is_kraft_mode() {
         result.insert(
             "zookeeper.connect".to_string(),
             "${env:ZOOKEEPER}".to_string(),
         );
     }
 
-    result.extend(kafka_security.controller_config_settings());
+    result.extend(cluster_config.kafka_security.controller_config_settings());
     result.extend(graceful_shutdown_config_properties());
     result.extend(overrides);
 

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

@@ -92,6 +92,20 @@ pub struct ValidatedClusterConfig {
     pub disable_broker_id_generation: bool,
 }
 
+impl ValidatedClusterConfig {
+    /// Whether the cluster runs in KRaft mode (as opposed to ZooKeeper mode).
+    pub fn is_kraft_mode(&self) -> bool {
+        self.metadata_manager == MetadataManager::KRaft
+    }
+
+    /// The OPA connect string, if OPA authorization is configured.
+    pub fn opa_connect(&self) -> Option<&str> {
+        self.authorization_config
+            .as_ref()
+            .map(|auth_config| auth_config.opa_connect.as_str())
+    }
+}
+
 /// Lets [`ValidatedCluster`] act as the owner [`Resource`] for child objects, so owner
 /// references are built from it (via the captured `metadata`) rather than the raw CR.
 impl Resource for ValidatedCluster {