imports, formatting

Author: adwk67

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

@@ -258,6 +258,17 @@ pub fn pod_fqdn(
     ))
 }
 
+pub fn pod_kcat(
+    kafka: &v1alpha1::KafkaCluster,
+    cluster_info: &KubernetesClusterInfo,
+) -> Result<String, KafkaListenerError> {
+    Ok(format!(
+        "$POD_NAME-listener-broker.{namespace}.svc.{cluster_domain}",
+        namespace = kafka.namespace().context(ObjectHasNoNamespaceSnafu)?,
+        cluster_domain = cluster_info.cluster_domain
+    ))
+}
+
 #[cfg(test)]
 mod tests {
     use stackable_operator::{

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

@@ -284,7 +284,7 @@ impl KafkaTlsSecurity {
     }
 
     /// Returns the commands for the kcat readiness probe.
-    pub fn kcat_prober_container_commands(&self, pod_fqdn: &String) -> Vec<String> {
+    pub fn kcat_prober_container_commands(&self, pod_kcat: &String) -> Vec<String> {
         let mut args = vec![];
         let port = self.client_port();
 
@@ -314,11 +314,11 @@ impl KafkaTlsSecurity {
             );
             bash_args.push("/stackable/kcat".to_string());
             bash_args.push("-b".to_string());
-            bash_args.push(format!("{pod_fqdn}:{port}"));
+            bash_args.push(format!("{pod_kcat}:{port}"));
             bash_args.extend(Self::kcat_client_sasl_ssl(
                 Self::STACKABLE_TLS_KCAT_DIR,
                 service_name,
-                pod_fqdn,
+                pod_kcat,
             ));
             bash_args.push("-L".to_string());
 
@@ -665,7 +665,7 @@ impl KafkaTlsSecurity {
     fn kcat_client_sasl_ssl(
         cert_directory: &str,
         service_name: &str,
-        pod_fqdn: &String,
+        pod_kcat: &String,
     ) -> Vec<String> {
         vec![
             "-X".to_string(),
@@ -679,7 +679,7 @@ impl KafkaTlsSecurity {
             "-X".to_string(),
             format!("sasl.kerberos.service.name={service_name}"),
             "-X".to_string(),
-            format!("sasl.kerberos.principal={service_name}/{pod_fqdn}@$KERBEROS_REALM"),
+            format!("sasl.kerberos.principal={service_name}/{pod_kcat}@$KERBEROS_REALM"),
         ]
     }
 }

rust/operator-binary/src/kafka_controller.rs

@@ -76,7 +76,7 @@ use strum::{EnumDiscriminants, IntoStaticStr};
 use crate::{
     config::jvm::{construct_heap_jvm_args, construct_non_heap_jvm_args},
     crd::{
-        listener::{get_kafka_listener_config, pod_fqdn, KafkaListenerError},
+        listener::{get_kafka_listener_config, pod_kcat, KafkaListenerError},
         security::KafkaTlsSecurity,
         v1alpha1, Container, KafkaClusterStatus, KafkaConfig, KafkaRole, APP_NAME,
         DOCKER_IMAGE_BASE_NAME, JVM_SECURITY_PROPERTIES_FILE, KAFKA_HEAP_OPTS,
@@ -999,8 +999,7 @@ fn build_broker_rolegroup_statefulset(
         .context(AddVolumeMountSnafu)?
         .resources(merged_config.resources.clone().into());
 
-    let pod_fqdn = pod_fqdn(kafka, &rolegroup_ref.object_name(), cluster_info)
-        .context(ResolveNamespaceSnafu)?;
+    let pod_kcat = pod_kcat(kafka, cluster_info).context(ResolveNamespaceSnafu)?;
     // Use kcat sidecar for probing container status rather than the official Kafka tools, since they incur a lot of
     // unacceptable perf overhead
     cb_kcat_prober
@@ -1030,7 +1029,7 @@ fn build_broker_rolegroup_statefulset(
         .readiness_probe(Probe {
             exec: Some(ExecAction {
                 // If the broker is able to get its fellow cluster members then it has at least completed basic registration at some point
-                command: Some(kafka_security.kcat_prober_container_commands(&pod_fqdn)),
+                command: Some(kafka_security.kcat_prober_container_commands(&pod_kcat)),
             }),
             timeout_seconds: Some(5),
             period_seconds: Some(2),