chore: move server properties from the command line to the configuration file

Author: razvan

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

@@ -10,13 +10,8 @@ use crate::{
     crd::{
         KafkaPodDescriptor, STACKABLE_CONFIG_DIR, STACKABLE_KERBEROS_KRB5_PATH,
         STACKABLE_LISTENER_BOOTSTRAP_DIR, STACKABLE_LISTENER_BROKER_DIR,
-        listener::{KafkaListenerConfig, KafkaListenerName, node_address_cmd},
-        role::{
-            KAFKA_ADVERTISED_LISTENERS, KAFKA_CONTROLLER_QUORUM_BOOTSTRAP_SERVERS,
-            KAFKA_CONTROLLER_QUORUM_VOTERS, KAFKA_LISTENER_SECURITY_PROTOCOL_MAP, KAFKA_LISTENERS,
-            KAFKA_NODE_ID, KAFKA_NODE_ID_OFFSET, KafkaRole, broker::BROKER_PROPERTIES_FILE,
-            controller::CONTROLLER_PROPERTIES_FILE,
-        },
+        listener::{KafkaListenerName, node_address_cmd},
+        role::{KafkaRole, broker::BROKER_PROPERTIES_FILE, controller::CONTROLLER_PROPERTIES_FILE},
         security::KafkaTlsSecurity,
         v1alpha1,
     },
@@ -28,8 +23,6 @@ pub fn broker_kafka_container_commands(
     kafka: &v1alpha1::KafkaCluster,
     cluster_id: &str,
     controller_descriptors: Vec<KafkaPodDescriptor>,
-    kafka_listeners: &KafkaListenerConfig,
-    opa_connect_string: Option<&str>,
     kafka_security: &KafkaTlsSecurity,
     product_version: &str,
 ) -> String {
@@ -51,26 +44,17 @@ pub fn broker_kafka_container_commands(
             true => format!("export KERBEROS_REALM=$(grep -oP 'default_realm = \\K.*' {STACKABLE_KERBEROS_KRB5_PATH})"),
             false => "".to_string(),
         },
-        broker_start_command = broker_start_command(kafka, cluster_id, controller_descriptors, kafka_listeners, opa_connect_string, kafka_security, product_version),
+        broker_start_command = broker_start_command(kafka, cluster_id, controller_descriptors, kafka_security, product_version),
     }
 }
 
 fn broker_start_command(
     kafka: &v1alpha1::KafkaCluster,
     cluster_id: &str,
     controller_descriptors: Vec<KafkaPodDescriptor>,
-    kafka_listeners: &KafkaListenerConfig,
-    opa_connect_string: Option<&str>,
     kafka_security: &KafkaTlsSecurity,
     product_version: &str,
 ) -> String {
-    let opa_config = match opa_connect_string {
-        None => "".to_string(),
-        Some(opa_connect_string) => {
-            format!(" --override \"opa.authorizer.url={opa_connect_string}\"")
-        }
-    };
-
     let jaas_config = match kafka_security.has_kerberos_enabled() {
         true => {
             formatdoc! {"
@@ -89,50 +73,35 @@ fn broker_start_command(
 
     let client_port = kafka_security.client_port();
 
-    // TODO: The properties file from the configmap is copied to the /tmp folder and appended with dynamic properties
     // This should be improved:
     // - mount emptyDir as readWriteConfig
-    // - use config-utils for proper replacements?
-    // - should we print the adapted properties file at startup?
     if kafka.is_controller_configured() {
         formatdoc! {"
-            export REPLICA_ID=$(echo \"$POD_NAME\" | grep -oE '[0-9]+$')
+            POD_INDEX=$(echo \"$POD_NAME\" | grep -oE '[0-9]+$')
+            export REPLICA_ID=$((POD_INDEX+NODE_ID_OFFSET))
+
             cp {config_dir}/{properties_file} /tmp/{properties_file}
 
-            echo \"{KAFKA_NODE_ID}=$((REPLICA_ID + ${KAFKA_NODE_ID_OFFSET}))\" >> /tmp/{properties_file}
-            echo \"{KAFKA_CONTROLLER_QUORUM_BOOTSTRAP_SERVERS}={bootstrap_servers}\" >> /tmp/{properties_file}
-            echo \"{KAFKA_LISTENERS}={listeners}\" >> /tmp/{properties_file}
-            echo \"{KAFKA_ADVERTISED_LISTENERS}={advertised_listeners}\" >> /tmp/{properties_file}
-            echo \"{KAFKA_LISTENER_SECURITY_PROTOCOL_MAP}={listener_security_protocol_map}\" >> /tmp/{properties_file}
-            echo \"{KAFKA_CONTROLLER_QUORUM_VOTERS}={controller_quorum_voters}\" >> /tmp/{properties_file}
+            config-utils template /tmp/{properties_file}
 
             bin/kafka-storage.sh format --cluster-id {cluster_id} --config /tmp/{properties_file} --ignore-formatted {initial_controller_command}
-            bin/kafka-server-start.sh /tmp/{properties_file} {opa_config}{jaas_config} &
+            bin/kafka-server-start.sh /tmp/{properties_file} {jaas_config} &
         ",
         config_dir = STACKABLE_CONFIG_DIR,
         properties_file = BROKER_PROPERTIES_FILE,
-        bootstrap_servers = to_bootstrap_servers(&controller_descriptors, client_port),
-        listeners = kafka_listeners.listeners(),
-        advertised_listeners = kafka_listeners.advertised_listeners(),
-        listener_security_protocol_map = kafka_listeners.listener_security_protocol_map(),
-        controller_quorum_voters = to_quorum_voters(&controller_descriptors, client_port),
         initial_controller_command = initial_controllers_command(&controller_descriptors, product_version, client_port),
         }
     } else {
         formatdoc! {"
-            bin/kafka-server-start.sh {config_dir}/{properties_file} \
-            --override \"zookeeper.connect=$ZOOKEEPER\" \
-            --override \"{KAFKA_LISTENERS}={listeners}\" \
-            --override \"{KAFKA_ADVERTISED_LISTENERS}={advertised_listeners}\" \
-            --override \"{KAFKA_LISTENER_SECURITY_PROTOCOL_MAP}={listener_security_protocol_map}\" \
-            {opa_config} \
+            cp {config_dir}/{properties_file} /tmp/{properties_file}
+
+            config-utils template /tmp/{properties_file}
+
+            bin/kafka-server-start.sh /tmp/{properties_file} \
             {jaas_config} \
             &",
         config_dir = STACKABLE_CONFIG_DIR,
         properties_file = BROKER_PROPERTIES_FILE,
-        listeners = kafka_listeners.listeners(),
-        advertised_listeners = kafka_listeners.advertised_listeners(),
-        listener_security_protocol_map = kafka_listeners.listener_security_protocol_map(),
         }
     }
 }
@@ -182,7 +151,6 @@ wait_for_termination()
 pub fn controller_kafka_container_command(
     cluster_id: &str,
     controller_descriptors: Vec<KafkaPodDescriptor>,
-    kafka_listeners: &KafkaListenerConfig,
     kafka_security: &KafkaTlsSecurity,
     product_version: &str,
 ) -> String {
@@ -199,14 +167,12 @@ pub fn controller_kafka_container_command(
         prepare_signal_handlers
         containerdebug --output={STACKABLE_LOG_DIR}/containerdebug-state.json --loop &
 
-        export REPLICA_ID=$(echo \"$POD_NAME\" | grep -oE '[0-9]+$')
+        POD_INDEX=$(echo \"$POD_NAME\" | grep -oE '[0-9]+$')
+        export REPLICA_ID=$((POD_INDEX+NODE_ID_OFFSET))
+
         cp {config_dir}/{properties_file} /tmp/{properties_file}
 
-        echo \"{KAFKA_NODE_ID}=$((REPLICA_ID + ${KAFKA_NODE_ID_OFFSET}))\" >> /tmp/{properties_file}
-        echo \"{KAFKA_CONTROLLER_QUORUM_BOOTSTRAP_SERVERS}={bootstrap_servers}\" >> /tmp/{properties_file}
-        echo \"{KAFKA_LISTENERS}={listeners}\" >> /tmp/{properties_file}
-        echo \"{KAFKA_LISTENER_SECURITY_PROTOCOL_MAP}={listener_security_protocol_map}\" >> /tmp/{properties_file}
-        echo \"{KAFKA_CONTROLLER_QUORUM_VOTERS}={controller_quorum_voters}\" >> /tmp/{properties_file}
+        config-utils template /tmp/{properties_file}
 
         bin/kafka-storage.sh format --cluster-id {cluster_id} --config /tmp/{properties_file} --ignore-formatted {initial_controller_command}
         bin/kafka-server-start.sh /tmp/{properties_file} &
@@ -217,29 +183,11 @@ pub fn controller_kafka_container_command(
         remove_vector_shutdown_file_command = remove_vector_shutdown_file_command(STACKABLE_LOG_DIR),
         config_dir = STACKABLE_CONFIG_DIR,
         properties_file = CONTROLLER_PROPERTIES_FILE,
-        bootstrap_servers = to_bootstrap_servers(&controller_descriptors, client_port),
-        listeners = to_listeners(client_port),
-        listener_security_protocol_map = to_listener_security_protocol_map(kafka_listeners),
         initial_controller_command = initial_controllers_command(&controller_descriptors, product_version, client_port),
-        controller_quorum_voters = to_quorum_voters(&controller_descriptors, client_port),
         create_vector_shutdown_file_command = create_vector_shutdown_file_command(STACKABLE_LOG_DIR)
     }
 }
 
-fn to_listeners(port: u16) -> String {
-    // The environment variables are set in the statefulset of the controller
-    format!(
-        "{listener_name}://$POD_NAME.$ROLEGROUP_HEADLESS_SERVICE_NAME.$NAMESPACE.svc.$CLUSTER_DOMAIN:{port}",
-        listener_name = KafkaListenerName::Controller
-    )
-}
-
-fn to_listener_security_protocol_map(kafka_listeners: &KafkaListenerConfig) -> String {
-    kafka_listeners
-        .listener_security_protocol_map_for_listener(&KafkaListenerName::Controller)
-        .unwrap_or("".to_string())
-}
-
 fn to_initial_controllers(controller_descriptors: &[KafkaPodDescriptor], port: u16) -> String {
     controller_descriptors
         .iter()
@@ -248,23 +196,6 @@ fn to_initial_controllers(controller_descriptors: &[KafkaPodDescriptor], port: u
         .join(",")
 }
 
-// TODO: This can be removed once 3.7.2 is removed. Used in command.rs.
-fn to_quorum_voters(controller_descriptors: &[KafkaPodDescriptor], port: u16) -> String {
-    controller_descriptors
-        .iter()
-        .map(|desc| desc.as_quorum_voter(port))
-        .collect::<Vec<String>>()
-        .join(",")
-}
-
-fn to_bootstrap_servers(controller_descriptors: &[KafkaPodDescriptor], port: u16) -> String {
-    controller_descriptors
-        .iter()
-        .map(|desc| format!("{fqdn}:{port}", fqdn = desc.fqdn()))
-        .collect::<Vec<String>>()
-        .join(",")
-}
-
 fn initial_controllers_command(
     controller_descriptors: &[KafkaPodDescriptor],
     product_version: &str,

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

@@ -2,6 +2,11 @@ use stackable_operator::role_utils::RoleGroupRef;
 
 use crate::crd::v1alpha1::KafkaCluster;
 
+/// The Kafka node.id needs to be unique across the Kafka cluster.
+/// This function generates an integer that is stable for a given role group
+/// regardless if broker or controllers.
+/// This integer is then added to the pod index to compute the final node.id
+/// TODO: this is dangerous. How high are the chances of these ranges overlapping?
 pub fn node_id_hash32_offset(rolegroup_ref: &RoleGroupRef<KafkaCluster>) -> u32 {
     let hash = fnv_hash32(&format!(
         "{role}-{rolegroup}",

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

@@ -331,11 +331,11 @@ pub fn get_kafka_listener_config(
 }
 
 pub fn node_address_cmd(directory: &str) -> String {
-    format!("$(cat {directory}/default-address/address)")
+    format!("${{file:UTF-8:{directory}/default-address/address}}")
 }
 
 pub fn node_port_cmd(directory: &str, port_name: &str) -> String {
-    format!("$(cat {directory}/default-address/ports/{port_name})")
+    format!("${{file:UTF-8:{directory}/default-address/ports/{port_name}}}")
 }
 
 pub fn pod_fqdn(
@@ -344,7 +344,7 @@ pub fn pod_fqdn(
     cluster_info: &KubernetesClusterInfo,
 ) -> Result<String, KafkaListenerError> {
     Ok(format!(
-        "$POD_NAME.{sts_service_name}.{namespace}.svc.{cluster_domain}",
+        "${{env:POD_NAME}}.{sts_service_name}.{namespace}.svc.{cluster_domain}",
         namespace = kafka.namespace().context(ObjectHasNoNamespaceSnafu)?,
         cluster_domain = cluster_info.cluster_domain
     ))

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

@@ -254,13 +254,14 @@ impl v1alpha1::KafkaCluster {
         })
     }
 
-    /// List all pod descriptors of a provided role expected to form the cluster.
+    /// List pod descriptors for a given role.
+    /// If no role is provided, pod descriptors for all roles are listed.
     ///
     /// We try to predict the pods here rather than looking at the current cluster state in order to
     /// avoid instance churn.
     pub fn pod_descriptors(
         &self,
-        requested_kafka_role: &KafkaRole,
+        requested_kafka_role: Option<&KafkaRole>,
         cluster_info: &KubernetesClusterInfo,
     ) -> Result<Vec<KafkaPodDescriptor>, Error> {
         let namespace = self.metadata.namespace.clone().context(NoNamespaceSnafu)?;
@@ -290,10 +291,12 @@ impl v1alpha1::KafkaCluster {
                 };
 
                 // only return descriptors for selected role
-                if current_role == *requested_kafka_role {
+                if requested_kafka_role.is_none() || &current_role == requested_kafka_role.unwrap()
+                {
                     for replica in 0..replicas {
                         pod_descriptors.push(KafkaPodDescriptor {
                             namespace: namespace.clone(),
+                            role: current_role.to_string(),
                             role_group_service_name: rolegroup_ref
                                 .rolegroup_headless_service_name(),
                             role_group_statefulset_name: rolegroup_ref.object_name(),
@@ -348,6 +351,7 @@ pub struct KafkaPodDescriptor {
     replica: u16,
     cluster_domain: DomainName,
     node_id: u32,
+    pub role: String,
 }
 
 impl KafkaPodDescriptor {

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

@@ -15,11 +15,7 @@ use stackable_operator::{
 use strum::{Display, EnumIter};
 
 use crate::crd::{
-    listener::KafkaListenerName,
-    role::{
-        KAFKA_LOG_DIRS, KAFKA_PROCESS_ROLES, KafkaRole,
-        commons::{CommonConfig, Storage, StorageFragment},
-    },
+    role::commons::{CommonConfig, Storage, StorageFragment},
     v1alpha1,
 };
 
@@ -121,29 +117,9 @@ impl Configuration for ControllerConfigFragment {
         &self,
         _resource: &Self::Configurable,
         _role_name: &str,
-        file: &str,
+        _file: &str,
     ) -> Result<BTreeMap<String, Option<String>>, stackable_operator::product_config_utils::Error>
     {
-        let mut config = BTreeMap::new();
-
-        if file == CONTROLLER_PROPERTIES_FILE {
-            config.insert(
-                KAFKA_LOG_DIRS.to_string(),
-                Some("/stackable/data/kraft".to_string()),
-            );
-
-            // KRAFT
-            config.insert(
-                KAFKA_PROCESS_ROLES.to_string(),
-                Some(KafkaRole::Controller.to_string()),
-            );
-
-            config.insert(
-                "controller.listener.names".to_string(),
-                Some(KafkaListenerName::Controller.to_string()),
-            );
-        }
-
-        Ok(config)
+        Ok(BTreeMap::new())
     }
 }

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

@@ -68,7 +68,8 @@ pub const KAFKA_CONTROLLER_QUORUM_BOOTSTRAP_SERVERS: &str = "controller.quorum.b
 
 /// Map of id/endpoint information for the set of voters in a comma-separated list of {id}@{host}:{port} entries.
 /// For example: 1@localhost:9092,2@localhost:9093,3@localhost:9094
-pub const KAFKA_CONTROLLER_QUORUM_VOTERS: &str = "controller.quorum.voters";
+/// TODO: maybe re-enable
+// pub const KAFKA_CONTROLLER_QUORUM_VOTERS: &str = "controller.quorum.voters";
 
 #[derive(Snafu, Debug)]
 pub enum Error {

rust/operator-binary/src/kafka_controller.rs

@@ -37,6 +37,7 @@ use crate::{
     crd::{
         self, APP_NAME, DOCKER_IMAGE_BASE_NAME, JVM_SECURITY_PROPERTIES_FILE, KafkaClusterStatus,
         OPERATOR_NAME,
+        listener::get_kafka_listener_config,
         role::{
             AnyConfig, KafkaRole, broker::BROKER_PROPERTIES_FILE,
             controller::CONTROLLER_PROPERTIES_FILE,
@@ -66,6 +67,14 @@ pub struct Ctx {
 #[strum_discriminants(derive(IntoStaticStr))]
 #[allow(clippy::enum_variant_names)]
 pub enum Error {
+    #[snafu(display("failed to build pod descriptors"))]
+    BuildPodDescriptors { source: crate::crd::Error },
+
+    #[snafu(display("invalid kafka listeners"))]
+    InvalidKafkaListeners {
+        source: crate::crd::listener::KafkaListenerError,
+    },
+
     #[snafu(display("cluster object defines no '{role}' role"))]
     MissingKafkaRole {
         source: crate::crd::Error,
@@ -242,6 +251,8 @@ impl ReconcilerError for Error {
             Error::BuildConfigMap { .. } => None,
             Error::BuildService { .. } => None,
             Error::BuildListener { .. } => None,
+            Error::InvalidKafkaListeners { .. } => None,
+            Error::BuildPodDescriptors { .. } => None,
         }
     }
 }
@@ -359,13 +370,28 @@ pub async fn reconcile_kafka(
                 build_rolegroup_metrics_service(kafka, &resolved_product_image, &rolegroup_ref)
                     .context(BuildServiceSnafu)?;
 
+            let kafka_listeners = get_kafka_listener_config(
+                kafka,
+                &kafka_security,
+                &rolegroup_ref,
+                &client.kubernetes_cluster_info,
+            )
+            .context(InvalidKafkaListenersSnafu)?;
+
+            let pod_descriptors = kafka
+                .pod_descriptors(None, &client.kubernetes_cluster_info)
+                .context(BuildPodDescriptorsSnafu)?;
+
             let rg_configmap = build_rolegroup_config_map(
                 kafka,
                 &resolved_product_image,
                 &kafka_security,
                 &rolegroup_ref,
                 rolegroup_config,
                 &merged_config,
+                &kafka_listeners,
+                &pod_descriptors,
+                opa_connect.as_deref(),
             )
             .context(BuildConfigMapSnafu)?;
 
@@ -376,7 +402,6 @@ pub async fn reconcile_kafka(
                     &resolved_product_image,
                     &rolegroup_ref,
                     rolegroup_config,
-                    opa_connect.as_deref(),
                     &kafka_security,
                     &merged_config,
                     &rbac_sa,