chore: Consolidate error enums and use expect where possible

CHANGELOG.md

@@ -25,6 +25,7 @@ All notable changes to this project will be documented in this file.
   Broker StatefulSets created by older operator versions cannot be updated in place: after the
   operator upgrade, delete each broker StatefulSet so that the operator immediately recreates it
   with the new labels ([#1011]).
+- Make operations infallible where dependent on static inputs ([#1017]).
 
 ### Fixed
 
@@ -41,6 +42,7 @@ All notable changes to this project will be documented in this file.
 [#1000]: https://github.com/stackabletech/kafka-operator/pull/1000
 [#1011]: https://github.com/stackabletech/kafka-operator/pull/1011
 [#1014]: https://github.com/stackabletech/kafka-operator/pull/1014
+[#1017]: https://github.com/stackabletech/kafka-operator/pull/1017
 
 ## [26.7.0] - 2026-07-21
 

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

@@ -15,7 +15,10 @@ use stackable_operator::{
     },
     commons::secret_class::SecretClassVolumeProvisionParts,
     constant,
-    v2::builder::pod::container::{EnvVarName, EnvVarSet},
+    v2::{
+        builder::pod::container::{EnvVarName, EnvVarSet},
+        types::kubernetes::VolumeName,
+    },
 };
 
 use crate::{
@@ -26,6 +29,8 @@ use crate::{
     },
 };
 
+constant!(KERBEROS_VOLUME_NAME: VolumeName = "kerberos");
+
 #[derive(Snafu, Debug)]
 pub enum Error {
     #[snafu(display("failed to add Kerberos secret volume"))]
@@ -35,13 +40,15 @@ pub enum Error {
 
     #[snafu(display("failed to add needed volume"))]
     AddVolume { source: builder::pod::Error },
-
-    #[snafu(display("failed to add needed volumeMount"))]
-    AddVolumeMount {
-        source: builder::pod::container::Error,
-    },
 }
 
+/// Adds the Kerberos keytab and `krb5.conf` volume to the pod builder and mounts it into the
+/// Kafka and kcat-prober containers, when Kerberos is enabled.
+///
+/// # Panics
+///
+/// Panics if the volume mounts cannot be added to the container builders. Only call this on
+/// container builders whose mount paths are still distinct from the ones added here.
 pub fn add_kerberos_pod_config(
     kafka_security: &ValidatedKafkaSecurity,
     role: &KafkaRole,
@@ -56,21 +63,23 @@ pub fn add_kerberos_pod_config(
             // We need both public (krb5.conf) and private (keytab) parts.
             SecretClassVolumeProvisionParts::PublicPrivate,
         )
-        .with_listener_volume_scope(LISTENER_BROKER_VOLUME_NAME)
-        .with_listener_volume_scope(LISTENER_BOOTSTRAP_VOLUME_NAME)
+        .with_listener_volume_scope(&*LISTENER_BROKER_VOLUME_NAME)
+        .with_listener_volume_scope(&*LISTENER_BOOTSTRAP_VOLUME_NAME)
         .with_kerberos_service_name(role.kerberos_service_name())
         .build()
         .context(KerberosSecretVolumeSnafu)?;
         pb.add_volume(
-            VolumeBuilder::new("kerberos")
+            VolumeBuilder::new(&*KERBEROS_VOLUME_NAME)
                 .ephemeral(kerberos_secret_operator_volume)
                 .build(),
         )
         .context(AddVolumeSnafu)?;
 
         for cb in [cb_kafka, cb_kcat_prober] {
-            cb.add_volume_mount("kerberos", STACKABLE_KERBEROS_DIR)
-                .context(AddVolumeMountSnafu)?;
+            cb.add_volume_mount(&*KERBEROS_VOLUME_NAME, STACKABLE_KERBEROS_DIR)
+                .expect(
+                    "The mount paths are statically defined and there should be no duplicates.",
+                );
         }
     }
 
@@ -108,5 +117,6 @@ mod tests {
         // Test that dereferencing the constants does not panic.
         let _ = *KRB5_CONFIG;
         let _ = *KAFKA_OPTS;
+        let _ = *KERBEROS_VOLUME_NAME;
     }
 }

rust/operator-binary/src/controller/build/resource/listener.rs

@@ -8,6 +8,7 @@ use stackable_operator::{
         role_group_utils::{QualifiedRoleGroupName, ResourceNames},
         types::{kubernetes::ListenerName, operator::ClusterName},
     },
+    validation::RFC_1035_LABEL_MAX_LENGTH,
 };
 
 use crate::{
@@ -24,26 +25,39 @@ use crate::{
 /// A free function (rather than only a [`ValidatedCluster`] method) so the dereference step can
 /// compute the name from the raw cluster identity when fetching the stored `Listener`s that the
 /// discovery `ConfigMap` is built from.
+///
+/// The returned name is both a valid [`ListenerName`] and a lowercase RFC 1035 label name. The
+/// length is ensured at compile time for both; the character class follows from
+/// [`QualifiedRoleGroupName`] being an RFC 1035 label name and is additionally checked by a unit
+/// test.
 pub fn bootstrap_listener_name(
     cluster_name: &ClusterName,
     role: &KafkaRole,
     role_group_name: &RoleGroupName,
 ) -> ListenerName {
     const BOOTSTRAP_SUFFIX: &str = "-bootstrap";
 
-    // Compile-time checks that `<qualified_role_group_name>-bootstrap` is a valid ListenerName, so
-    // the `expect` below cannot fire.
+    // Compile-time checks that `<qualified_role_group_name>-bootstrap` is both a valid ListenerName
+    // and an RFC 1035 label name, so the `expect` below cannot fire.
     //
     // Length: the qualified role group name plus the suffix stays within the ListenerName limit.
     const _: () = assert!(
         QualifiedRoleGroupName::MAX_LENGTH + BOOTSTRAP_SUFFIX.len() <= ListenerName::MAX_LENGTH,
         "The string `<qualified_role_group_name>-bootstrap` must not exceed the limit of Listener \
-    names."
+        names."
+    );
+    // Length: the qualified role group name plus the suffix stays within the RFC 1035 label limit.
+    const _: () = assert!(
+        QualifiedRoleGroupName::MAX_LENGTH + BOOTSTRAP_SUFFIX.len() <= RFC_1035_LABEL_MAX_LENGTH,
+        "The string `<qualified_role_group_name>-bootstrap` must not exceed the limit of an \
+        RFC 1035 label name."
     );
-    // Characters: a ListenerName is an RFC 1123 DNS subdomain. The qualified role group name is an
-    // RFC 1123 label name (which is a subdomain of a single label); appending `-bootstrap` keeps it
-    // one, as the name still starts and ends with an alphanumeric character and adds no invalid ones.
+    // Characters: the qualified role group name is an RFC 1123 DNS subdomain name (which a
+    // ListenerName requires) and an RFC 1035 label name. Appending `-bootstrap` adds only lowercase
+    // letters and a dash and ends with a letter, so the result is still both.
     let _ = QualifiedRoleGroupName::IS_RFC_1123_SUBDOMAIN_NAME;
+    let _ = QualifiedRoleGroupName::IS_RFC_1035_LABEL_NAME;
+    let _ = ListenerName::IS_RFC_1123_SUBDOMAIN_NAME;
 
     let resource_names = ResourceNames {
         cluster_name: cluster_name.clone(),
@@ -58,8 +72,8 @@ pub fn bootstrap_listener_name(
     .expect("is a valid Listener name")
 }
 
-/// Kafka clients will use the load-balanced bootstrap listener to get a list of broker addresses and will use those to
-/// transmit data to the correct broker.
+/// Kafka clients will use the load-balanced bootstrap listener to get a list of broker addresses
+/// and will use those to transmit data to the correct broker.
 // TODO (@NickLarsenNZ): Move shared functionality to stackable-operator
 pub fn build_broker_rolegroup_bootstrap_listener(
     validated_cluster: &ValidatedCluster,
@@ -110,3 +124,35 @@ fn bootstrap_listener_ports(
         }
     }]
 }
+
+#[cfg(test)]
+mod tests {
+    use stackable_operator::validation::RFC_1123_LABEL_MAX_LENGTH;
+    use strum::IntoEnumIterator;
+
+    use super::*;
+
+    #[test]
+    fn bootstrap_listener_name_is_rfc_1035_label_name() {
+        // The length is already ensured at compile time; this test covers the character class.
+        // Every ClusterName is a valid RFC 1035 label name, so we use just some string with maximum
+        // length. The role group name is user-provided, so use the maximum length of an RFC 1123
+        // label there as well; operator-rs then hash-truncates the qualified role group name.
+        let _ = ClusterName::IS_RFC_1035_LABEL_NAME;
+        let cluster_name = ClusterName::from_str(&"a".repeat(ClusterName::MAX_LENGTH))
+            .expect("is a valid ClusterName");
+        let role_group_name = RoleGroupName::from_str(&"g".repeat(RFC_1123_LABEL_MAX_LENGTH))
+            .expect("is a valid RoleGroupName");
+
+        for role in KafkaRole::iter() {
+            let bootstrap_listener_name =
+                bootstrap_listener_name(&cluster_name, &role, &role_group_name);
+            assert!(
+                stackable_operator::validation::is_lowercase_rfc_1035_label(
+                    bootstrap_listener_name.as_ref()
+                )
+                .is_ok()
+            );
+        }
+    }
+}