fix 3.7 deployments and tests

Author: razvan

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

@@ -71,8 +71,6 @@ fn broker_start_command(
         false => "".to_string(),
     };
 
-    let client_port = kafka_security.client_port();
-
     // This should be improved:
     // - mount emptyDir as readWriteConfig
     if kafka.is_controller_configured() {
@@ -89,7 +87,7 @@ fn broker_start_command(
         ",
         config_dir = STACKABLE_CONFIG_DIR,
         properties_file = BROKER_PROPERTIES_FILE,
-        initial_controller_command = initial_controllers_command(&controller_descriptors, product_version, client_port),
+        initial_controller_command = initial_controllers_command(&controller_descriptors, product_version),
         }
     } else {
         formatdoc! {"
@@ -151,11 +149,8 @@ wait_for_termination()
 pub fn controller_kafka_container_command(
     cluster_id: &str,
     controller_descriptors: Vec<KafkaPodDescriptor>,
-    kafka_security: &KafkaTlsSecurity,
     product_version: &str,
 ) -> String {
-    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
@@ -183,29 +178,28 @@ 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,
-        initial_controller_command = initial_controllers_command(&controller_descriptors, product_version, client_port),
+        initial_controller_command = initial_controllers_command(&controller_descriptors, product_version),
         create_vector_shutdown_file_command = create_vector_shutdown_file_command(STACKABLE_LOG_DIR)
     }
 }
 
-fn to_initial_controllers(controller_descriptors: &[KafkaPodDescriptor], port: u16) -> String {
+fn to_initial_controllers(controller_descriptors: &[KafkaPodDescriptor]) -> String {
     controller_descriptors
         .iter()
-        .map(|desc| desc.as_voter(port))
+        .map(|desc| desc.as_voter())
         .collect::<Vec<String>>()
         .join(",")
 }
 
 fn initial_controllers_command(
     controller_descriptors: &[KafkaPodDescriptor],
     product_version: &str,
-    client_port: u16,
 ) -> String {
     match product_version.starts_with("3.7") {
         true => "".to_string(),
         false => format!(
             "--initial-controllers {initial_controllers}",
-            initial_controllers = to_initial_controllers(controller_descriptors, client_port),
+            initial_controllers = to_initial_controllers(controller_descriptors),
         ),
     }
 }

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

@@ -6,7 +6,8 @@ use crate::crd::v1alpha1::KafkaCluster;
 /// 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?
+/// The node.id is only set and used in Kraft mode.
+/// Warning: this is not safe from collisions.
 pub fn node_id_hash32_offset(rolegroup_ref: &RoleGroupRef<KafkaCluster>) -> u32 {
     let hash = fnv_hash32(&format!(
         "{role}-{rolegroup}",

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

@@ -263,6 +263,7 @@ impl v1alpha1::KafkaCluster {
         &self,
         requested_kafka_role: Option<&KafkaRole>,
         cluster_info: &KubernetesClusterInfo,
+        client_port: u16,
     ) -> Result<Vec<KafkaPodDescriptor>, Error> {
         let namespace = self.metadata.namespace.clone().context(NoNamespaceSnafu)?;
         let mut pod_descriptors = Vec::new();
@@ -303,6 +304,7 @@ impl v1alpha1::KafkaCluster {
                             replica,
                             cluster_domain: cluster_info.cluster_domain.clone(),
                             node_id: node_id_hash_offset + u32::from(replica),
+                            client_port,
                         });
                     }
                 }
@@ -352,6 +354,7 @@ pub struct KafkaPodDescriptor {
     cluster_domain: DomainName,
     node_id: u32,
     pub role: String,
+    pub client_port: u16,
 }
 
 impl KafkaPodDescriptor {
@@ -380,18 +383,20 @@ impl KafkaPodDescriptor {
     ///   * controller-0 is the replica's host,
     ///   * 1234 is the replica's port.
     // NOTE(@maltesander): Even though the used Uuid states to be type 4 it does not work... 0000000000-00000000000 works...
-    pub fn as_voter(&self, port: u16) -> String {
+    pub fn as_voter(&self) -> String {
         format!(
             "{node_id}@{fqdn}:{port}:0000000000-{node_id:0>11}",
             node_id = self.node_id,
+            port = self.client_port,
             fqdn = self.fqdn(),
         )
     }
 
-    pub fn as_quorum_voter(&self, port: u16) -> String {
+    pub fn as_quorum_voter(&self) -> String {
         format!(
             "{node_id}@{fqdn}:{port}",
             node_id = self.node_id,
+            port = self.client_port,
             fqdn = self.fqdn(),
         )
     }

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

@@ -68,8 +68,7 @@ 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
-/// TODO: maybe re-enable
-// pub const KAFKA_CONTROLLER_QUORUM_VOTERS: &str = "controller.quorum.voters";
+pub const KAFKA_CONTROLLER_QUORUM_VOTERS: &str = "controller.quorum.voters";
 
 #[derive(Snafu, Debug)]
 pub enum Error {

rust/operator-binary/src/kafka_controller.rs

@@ -379,7 +379,11 @@ pub async fn reconcile_kafka(
             .context(InvalidKafkaListenersSnafu)?;
 
             let pod_descriptors = kafka
-                .pod_descriptors(None, &client.kubernetes_cluster_info)
+                .pod_descriptors(
+                    None,
+                    &client.kubernetes_cluster_info,
+                    kafka_security.client_port(),
+                )
                 .context(BuildPodDescriptorsSnafu)?;
 
             let rg_configmap = build_rolegroup_config_map(

rust/operator-binary/src/resource/configmap.rs

@@ -18,8 +18,8 @@ use crate::{
         listener::{KafkaListenerConfig, KafkaListenerName},
         role::{
             AnyConfig, KAFKA_ADVERTISED_LISTENERS, KAFKA_CONTROLLER_QUORUM_BOOTSTRAP_SERVERS,
-            KAFKA_LISTENER_SECURITY_PROTOCOL_MAP, KAFKA_LISTENERS, KAFKA_LOG_DIRS, KAFKA_NODE_ID,
-            KAFKA_PROCESS_ROLES, KafkaRole,
+            KAFKA_CONTROLLER_QUORUM_VOTERS, KAFKA_LISTENER_SECURITY_PROTOCOL_MAP, KAFKA_LISTENERS,
+            KAFKA_LOG_DIRS, KAFKA_NODE_ID, KAFKA_PROCESS_ROLES, KafkaRole,
         },
         security::KafkaTlsSecurity,
         v1alpha1,
@@ -94,6 +94,7 @@ pub fn build_rolegroup_config_map(
         pod_descriptors,
         listener_config,
         opa_connect_string,
+        resolved_product_image.product_version.starts_with("3.7"), // needs_quorum_voters
     )?;
 
     // Need to call this to get configOverrides :(
@@ -198,6 +199,7 @@ fn server_properties_file(
     pod_descriptors: &[KafkaPodDescriptor],
     listener_config: &KafkaListenerConfig,
     opa_connect_string: Option<&str>,
+    needs_quorum_voters: bool,
 ) -> Result<BTreeMap<String, String>, Error> {
     let kraft_controllers = kraft_controllers(pod_descriptors);
 
@@ -209,7 +211,7 @@ fn server_properties_file(
         KafkaRole::Controller => {
             let kraft_controllers = kraft_controllers.context(NoKraftControllersFoundSnafu)?;
 
-            Ok(BTreeMap::from([
+            let mut result = BTreeMap::from([
             (
                 KAFKA_LOG_DIRS.to_string(),
                 "/stackable/data/kraft".to_string(),
@@ -227,10 +229,6 @@ fn server_properties_file(
                 KAFKA_CONTROLLER_QUORUM_BOOTSTRAP_SERVERS.to_string(),
                 kraft_controllers.clone(),
             ),
-            // TODO: figure this out
-            //(KAFKA_CONTROLLER_QUORUM_VOTERS.to_string(),
-            //kraft_controllers,
-            //),
             (
                 KAFKA_LISTENERS.to_string(),
                 "CONTROLLER://${env:POD_NAME}.${env:ROLEGROUP_HEADLESS_SERVICE_NAME}.${env:NAMESPACE}.svc.${env:CLUSTER_DOMAIN}:${env:KAFKA_CLIENT_PORT}".to_string(),
@@ -240,7 +238,16 @@ fn server_properties_file(
                 listener_config
                     .listener_security_protocol_map_for_listener(&KafkaListenerName::Controller)
                     .unwrap_or("".to_string())),
-            ]))
+            ]);
+
+            if needs_quorum_voters {
+                let kraft_voters =
+                    kraft_voters(pod_descriptors).context(NoKraftControllersFoundSnafu)?;
+
+                result.extend([(KAFKA_CONTROLLER_QUORUM_VOTERS.to_string(), kraft_voters)]);
+            }
+
+            Ok(result)
         }
         KafkaRole::Broker => {
             let mut result = BTreeMap::from([
@@ -278,6 +285,13 @@ fn server_properties_file(
                         kraft_controllers.clone(),
                     ),
                 ]);
+
+                if needs_quorum_voters {
+                    let kraft_voters =
+                        kraft_voters(pod_descriptors).context(NoKraftControllersFoundSnafu)?;
+
+                    result.extend([(KAFKA_CONTROLLER_QUORUM_VOTERS.to_string(), kraft_voters)]);
+                }
             } else {
                 // Running with ZooKeeper enabled
                 result.extend([(
@@ -313,7 +327,28 @@ fn kraft_controllers(pod_descriptors: &[KafkaPodDescriptor]) -> Option<String> {
     let result = pod_descriptors
         .iter()
         .filter(|pd| pd.role == KafkaRole::Controller.to_string())
-        .map(|desc| format!("{fqdn}:${{env:KAFKA_CLIENT_PORT}}", fqdn = desc.fqdn()))
+        .map(|desc| {
+            format!(
+                "{fqdn}:{client_port}",
+                fqdn = desc.fqdn(),
+                client_port = desc.client_port
+            )
+        })
+        .collect::<Vec<String>>()
+        .join(",");
+
+    if result.is_empty() {
+        None
+    } else {
+        Some(result)
+    }
+}
+
+fn kraft_voters(pod_descriptors: &[KafkaPodDescriptor]) -> Option<String> {
+    let result = pod_descriptors
+        .iter()
+        .filter(|pd| pd.role == KafkaRole::Controller.to_string())
+        .map(|desc| desc.as_quorum_voter())
         .collect::<Vec<String>>()
         .join(",");
 

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

@@ -299,7 +299,11 @@ pub fn build_broker_rolegroup_statefulset(
             cluster_id,
             // we need controller pods
             kafka
-                .pod_descriptors(Some(&KafkaRole::Controller), cluster_info)
+                .pod_descriptors(
+                    Some(&KafkaRole::Controller),
+                    cluster_info,
+                    kafka_security.client_port(),
+                )
                 .context(BuildPodDescriptorsSnafu)?,
             kafka_security,
             &resolved_product_image.product_version,
@@ -642,9 +646,8 @@ pub fn build_controller_rolegroup_statefulset(
         .args(vec![controller_kafka_container_command(
             kafka.cluster_id().context(ClusterIdMissingSnafu)?,
             kafka
-                .pod_descriptors(Some(kafka_role), cluster_info)
+                .pod_descriptors(Some(kafka_role), cluster_info, kafka_security.client_port())
                 .context(BuildPodDescriptorsSnafu)?,
-            kafka_security,
             &resolved_product_image.product_version,
         )])
         .add_env_var("PRE_STOP_CONTROLLER_SLEEP_SECONDS", "10")