refactor: use v2 Port

Author: maltesander

rust/operator-binary/src/controller.rs

@@ -191,20 +191,18 @@ impl ValidatedCluster {
 
                 if requested_kafka_role.is_none() || Some(role) == requested_kafka_role {
                     let resource_names = self.resource_names(role, role_group_name);
-                    let role_group_statefulset_name =
-                        resource_names.stateful_set_name().to_string();
-                    let role_group_service_name =
-                        resource_names.headless_service_name().to_string();
+                    let role_group_statefulset_name = resource_names.stateful_set_name();
+                    let role_group_service_name = resource_names.headless_service_name();
                     for replica in 0..validated_rg.replicas {
                         pod_descriptors.push(KafkaPodDescriptor {
-                            namespace: self.namespace.to_string(),
-                            role: role.to_string(),
+                            namespace: self.namespace.clone(),
                             role_group_service_name: role_group_service_name.clone(),
                             role_group_statefulset_name: role_group_statefulset_name.clone(),
                             replica,
                             cluster_domain: self.cluster_domain.clone(),
                             node_id: node_id_hash_offset + u32::from(replica),
-                            client_port,
+                            role: role.clone(),
+                            client_port: client_port.clone(),
                         });
                     }
                 }

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

@@ -59,7 +59,7 @@ pub fn uses_legacy_log4j(product_version: &str) -> bool {
 pub(crate) fn kraft_controllers(pod_descriptors: &[KafkaPodDescriptor]) -> Vec<String> {
     pod_descriptors
         .iter()
-        .filter(|pd| pd.role == KafkaRole::Controller.to_string())
+        .filter(|pd| pd.role == KafkaRole::Controller)
         .map(|desc| {
             format!(
                 "{fqdn}:{client_port}",

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

@@ -21,8 +21,12 @@ use stackable_operator::{
     schemars::{self, JsonSchema},
     status::condition::{ClusterCondition, HasStatusCondition},
     v2::{
-        config_overrides::KeyValueConfigOverrides, role_utils::JavaCommonConfig,
-        types::kubernetes::ConfigMapName,
+        config_overrides::KeyValueConfigOverrides,
+        role_utils::JavaCommonConfig,
+        types::{
+            common::Port,
+            kubernetes::{ConfigMapName, NamespaceName, ServiceName, StatefulSetName},
+        },
     },
     versioned::versioned,
 };
@@ -40,7 +44,7 @@ pub const OPERATOR_NAME: &str = "kafka.stackable.tech";
 pub const FIELD_MANAGER: &str = "kafka-operator";
 // metrics
 pub const METRICS_PORT_NAME: &str = "metrics";
-pub const METRICS_PORT: u16 = 9606;
+pub const METRICS_PORT: Port = Port(9606);
 // env vars
 pub const KAFKA_HEAP_OPTS: &str = "KAFKA_HEAP_OPTS";
 // server_properties
@@ -350,16 +354,16 @@ impl v1alpha1::KafkaCluster {
 /// Reference to a single `Pod` that is a component of a [`KafkaCluster`]
 ///
 /// Used for service discovery.
-#[derive(Debug, PartialEq, Eq, PartialOrd, Ord)]
+#[derive(Debug, PartialEq, Eq)]
 pub struct KafkaPodDescriptor {
-    pub(crate) namespace: String,
-    pub(crate) role_group_statefulset_name: String,
-    pub(crate) role_group_service_name: String,
+    pub(crate) namespace: NamespaceName,
+    pub(crate) role_group_statefulset_name: StatefulSetName,
+    pub(crate) role_group_service_name: ServiceName,
     pub(crate) replica: u16,
     pub(crate) cluster_domain: DomainName,
     pub(crate) node_id: u32,
-    pub role: String,
-    pub client_port: u16,
+    pub role: KafkaRole,
+    pub client_port: Port,
 }
 
 impl KafkaPodDescriptor {

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

@@ -20,7 +20,7 @@ use stackable_operator::{
     crd::authentication::core,
     k8s_openapi::api::core::v1::Volume,
     shared::time::Duration,
-    v2::types::kubernetes::SecretClassName,
+    v2::types::{common::Port, kubernetes::SecretClassName},
 };
 
 use super::listener::KafkaListenerProtocol;
@@ -66,28 +66,28 @@ pub struct KafkaTlsSecurity {
 }
 
 impl KafkaTlsSecurity {
-    pub const BOOTSTRAP_PORT: u16 = 9094;
+    pub const BOOTSTRAP_PORT: Port = Port(9094);
     // bootstrap: we will have a single named port with different values for
     // secure (9095) and insecure (9094). The bootstrap listener is needed to
     // be able to expose principals for both the broker and bootstrap in the
     // JAAS configuration, so that clients can use both.
     pub const BOOTSTRAP_PORT_NAME: &'static str = "bootstrap";
-    pub const CLIENT_PORT: u16 = 9092;
+    pub const CLIENT_PORT: Port = Port(9092);
     // ports
     pub const CLIENT_PORT_NAME: &'static str = "kafka";
     // internal
-    pub const INTERNAL_PORT: u16 = 19092;
+    pub const INTERNAL_PORT: Port = Port(19092);
     // - TLS internal
     const INTER_BROKER_LISTENER_NAME: &'static str = "inter.broker.listener.name";
     // - TLS global
     const KEYSTORE_P12_FILE_NAME: &'static str = "keystore.p12";
     const OPA_TLS_MOUNT_PATH: &str = "/stackable/tls-opa";
     // opa
     const OPA_TLS_VOLUME_NAME: &str = "tls-opa";
-    pub const SECURE_BOOTSTRAP_PORT: u16 = 9095;
-    pub const SECURE_CLIENT_PORT: u16 = 9093;
+    pub const SECURE_BOOTSTRAP_PORT: Port = Port(9095);
+    pub const SECURE_CLIENT_PORT: Port = Port(9093);
     pub const SECURE_CLIENT_PORT_NAME: &'static str = "kafka-tls";
-    pub const SECURE_INTERNAL_PORT: u16 = 19093;
+    pub const SECURE_INTERNAL_PORT: Port = Port(19093);
     const SSL_STORE_PASSWORD: &'static str = "";
     const SSL_STORE_TYPE_PKCS12: &'static str = "PKCS12";
     const STACKABLE_TLS_KAFKA_INTERNAL_DIR: &'static str = "/stackable/tls-kafka-internal";
@@ -222,15 +222,15 @@ impl KafkaTlsSecurity {
     }
 
     /// Return the Kafka (secure) client port depending on tls or authentication settings.
-    pub fn client_port(&self) -> u16 {
+    pub fn client_port(&self) -> Port {
         if self.tls_enabled() {
             Self::SECURE_CLIENT_PORT
         } else {
             Self::CLIENT_PORT
         }
     }
 
-    pub fn bootstrap_port(&self) -> u16 {
+    pub fn bootstrap_port(&self) -> Port {
         if self.tls_enabled() {
             Self::SECURE_BOOTSTRAP_PORT
         } else {
@@ -252,7 +252,7 @@ impl KafkaTlsSecurity {
     }
 
     /// Return the Kafka (secure) internal port depending on tls settings.
-    pub fn internal_port(&self) -> u16 {
+    pub fn internal_port(&self) -> Port {
         if self.tls_internal_secret_class().is_some() {
             Self::SECURE_INTERNAL_PORT
         } else {