stackabletech · NickLarsenNZ · Apr 3, 2025 · Apr 2, 2025 · Apr 2, 2025 · Apr 2, 2025
diff --git a/.github/PULL_REQUEST_TEMPLATE/pre-release-rust-deps.md b/.github/PULL_REQUEST_TEMPLATE/pre-release-rust-deps.md
@@ -1,4 +1,4 @@
-# Bump Rust Dependencies for Stackable Release XX.(X)X
+# Bump Rust Dependencies for Stackable Release YY.M.X
 
 <!--
     Make sure to update the link in 'issues/.github/ISSUE_TEMPLATE/pre-release-operator-rust-deps.md'
@@ -32,7 +32,7 @@ Part of <https://github.com/stackabletech/issues/issues/TRACKING_ISSUE>
 
 ```[tasklist]
 ### Bump Rust Dependencies
-- [ ] Bump `stackable-operator` and friends.
-- [ ] Bump `product-version`.
-- [ ] Bump all other dependencies.
+- [ ] Bump `stackable-operator` and friends
+- [ ] Bump `product-config`
+- [ ] Bump all other dependencies
 ```
diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml
@@ -26,7 +26,8 @@ env:
   CARGO_TERM_COLOR: always
   CARGO_INCREMENTAL: '0'
   CARGO_PROFILE_DEV_DEBUG: '0'
-  RUST_TOOLCHAIN_VERSION: "1.84.1"
+  RUST_TOOLCHAIN_VERSION: "1.85.0"
+  RUST_NIGHTLY_TOOLCHAIN_VERSION: "nightly-2025-01-15"
   PYTHON_VERSION: "3.12"
   RUSTFLAGS: "-D warnings"
   RUSTDOCFLAGS: "-D warnings"
@@ -136,9 +137,11 @@ jobs:
           submodules: recursive
       - uses: dtolnay/rust-toolchain@c5a29ddb4d9d194e7c84ec8c3fba61b1c31fee8c
         with:
-          toolchain: ${{ env.RUST_TOOLCHAIN_VERSION }}
+          toolchain: ${{ env.RUST_NIGHTLY_TOOLCHAIN_VERSION }}
           components: rustfmt
-      - run: cargo fmt --all -- --check
+      - env:
+          RUST_TOOLCHAIN_VERSION: ${{ env.RUST_NIGHTLY_TOOLCHAIN_VERSION }}
+        run: cargo "+$RUST_TOOLCHAIN_VERSION" fmt --all -- --check
 
   run_clippy:
     name: Run Clippy

diff --git a/.vscode/settings.json b/.vscode/settings.json
@@ -0,0 +1,7 @@
+{
+    "rust-analyzer.rustfmt.overrideCommand": [
+        "rustfmt",
+        "+nightly-2025-01-15",
+        "--"
+    ],
+}
diff --git a/rust-toolchain.toml b/rust-toolchain.toml
@@ -1,3 +1,3 @@
 # DO NOT EDIT, this file is generated by operator-templating
 [toolchain]
-channel = "1.84.1"
+channel = "1.85.0"
diff --git a/rust/operator-binary/src/config/jvm.rs b/rust/operator-binary/src/config/jvm.rs
@@ -5,7 +5,7 @@ use stackable_operator::{
 };
 
 use crate::crd::{
-    KafkaConfig, KafkaConfigFragment, JVM_SECURITY_PROPERTIES_FILE, METRICS_PORT,
+    JVM_SECURITY_PROPERTIES_FILE, KafkaConfig, KafkaConfigFragment, METRICS_PORT,
     STACKABLE_CONFIG_DIR,
 };
 
@@ -51,7 +51,9 @@ fn construct_jvm_args(
         format!("-Xmx{java_heap}"),
         format!("-Xms{java_heap}"),
         format!("-Djava.security.properties={STACKABLE_CONFIG_DIR}/{JVM_SECURITY_PROPERTIES_FILE}"),
-        format!("-javaagent:/stackable/jmx/jmx_prometheus_javaagent.jar={METRICS_PORT}:/stackable/jmx/broker.yaml")
+        format!(
+            "-javaagent:/stackable/jmx/jmx_prometheus_javaagent.jar={METRICS_PORT}:/stackable/jmx/broker.yaml"
+        ),
     ];
 
     let operator_generated = JvmArgumentOverrides::new_with_only_additions(jvm_args);
@@ -99,7 +101,7 @@ fn is_heap_jvm_argument(jvm_argument: &str) -> bool {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::crd::{v1alpha1, KafkaRole};
+    use crate::crd::{KafkaRole, v1alpha1};
 
     #[test]
     fn test_construct_jvm_arguments_defaults() {

diff --git a/rust/operator-binary/src/crd/affinity.rs b/rust/operator-binary/src/crd/affinity.rs
@@ -1,9 +1,9 @@
 use stackable_operator::{
-    commons::affinity::{affinity_between_role_pods, StackableAffinityFragment},
+    commons::affinity::{StackableAffinityFragment, affinity_between_role_pods},
     k8s_openapi::api::core::v1::PodAntiAffinity,
 };
 
-use crate::crd::{KafkaRole, APP_NAME};
+use crate::crd::{APP_NAME, KafkaRole};
 
 pub fn get_affinity(cluster_name: &str, role: &KafkaRole) -> StackableAffinityFragment {
     StackableAffinityFragment {
@@ -60,39 +60,36 @@ mod tests {
             .merged_config(&role, &role.rolegroup_ref(&kafka, "default"))
             .unwrap();
 
-        assert_eq!(
-            merged_config.affinity,
-            StackableAffinity {
-                pod_affinity: None,
-                pod_anti_affinity: Some(PodAntiAffinity {
-                    preferred_during_scheduling_ignored_during_execution: Some(vec![
-                        WeightedPodAffinityTerm {
-                            pod_affinity_term: PodAffinityTerm {
-                                label_selector: Some(LabelSelector {
-                                    match_expressions: None,
-                                    match_labels: Some(BTreeMap::from([
-                                        ("app.kubernetes.io/name".to_string(), "kafka".to_string(),),
-                                        (
-                                            "app.kubernetes.io/instance".to_string(),
-                                            "simple-kafka".to_string(),
-                                        ),
-                                        (
-                                            "app.kubernetes.io/component".to_string(),
-                                            "broker".to_string(),
-                                        )
-                                    ]))
-                                }),
-                                topology_key: "kubernetes.io/hostname".to_string(),
-                                ..PodAffinityTerm::default()
-                            },
-                            weight: 70
-                        }
-                    ]),
-                    required_during_scheduling_ignored_during_execution: None,
-                }),
-                node_affinity: None,
-                node_selector: None,
-            }
-        );
+        assert_eq!(merged_config.affinity, StackableAffinity {
+            pod_affinity: None,
+            pod_anti_affinity: Some(PodAntiAffinity {
+                preferred_during_scheduling_ignored_during_execution: Some(vec![
+                    WeightedPodAffinityTerm {
+                        pod_affinity_term: PodAffinityTerm {
+                            label_selector: Some(LabelSelector {
+                                match_expressions: None,
+                                match_labels: Some(BTreeMap::from([
+                                    ("app.kubernetes.io/name".to_string(), "kafka".to_string(),),
+                                    (
+                                        "app.kubernetes.io/instance".to_string(),
+                                        "simple-kafka".to_string(),
+                                    ),
+                                    (
+                                        "app.kubernetes.io/component".to_string(),
+                                        "broker".to_string(),
+                                    )
+                                ]))
+                            }),
+                            topology_key: "kubernetes.io/hostname".to_string(),
+                            ..PodAffinityTerm::default()
+                        },
+                        weight: 70
+                    }
+                ]),
+                required_during_scheduling_ignored_during_execution: None,
+            }),
+            node_affinity: None,
+            node_selector: None,
+        });
     }
 }
diff --git a/rust/operator-binary/src/crd/authentication.rs b/rust/operator-binary/src/crd/authentication.rs
@@ -18,7 +18,9 @@ pub enum Error {
         authentication_class: ObjectRef<AuthenticationClass>,
     },
 
-    #[snafu(display("only one authentication class at a time is currently supported. Possible Authentication class providers are {SUPPORTED_AUTHENTICATION_CLASS_PROVIDERS:?}"))]
+    #[snafu(display(
+        "only one authentication class at a time is currently supported. Possible Authentication class providers are {SUPPORTED_AUTHENTICATION_CLASS_PROVIDERS:?}"
+    ))]
     MultipleAuthenticationClassesProvided,
 
     #[snafu(display(
@@ -122,7 +124,7 @@ impl ResolvedAuthenticationClasses {
                     return Err(Error::AuthenticationProviderNotSupported {
                         authentication_class: ObjectRef::from_obj(auth_class),
                         provider: auth_class.spec.provider.to_string(),
-                    })
+                    });
                 }
             }
         }

diff --git a/rust/operator-binary/src/crd/listener.rs b/rust/operator-binary/src/crd/listener.rs
@@ -7,7 +7,7 @@ use snafu::{OptionExt, Snafu};
 use stackable_operator::{kube::ResourceExt, utils::cluster_info::KubernetesClusterInfo};
 use strum::{EnumDiscriminants, EnumString};
 
-use crate::crd::{security::KafkaTlsSecurity, v1alpha1, STACKABLE_LISTENER_BROKER_DIR};
+use crate::crd::{STACKABLE_LISTENER_BROKER_DIR, security::KafkaTlsSecurity, v1alpha1};
 
 const LISTENER_LOCAL_ADDRESS: &str = "0.0.0.0";
 
@@ -264,9 +264,9 @@ mod tests {
         builder::meta::ObjectMetaBuilder,
         commons::{
             authentication::{
+                AuthenticationClass, AuthenticationClassProvider, AuthenticationClassSpec,
                 kerberos,
                 tls::{self},
-                AuthenticationClass, AuthenticationClassProvider, AuthenticationClassSpec,
             },
             networking::DomainName,
         },
@@ -512,7 +512,7 @@ mod tests {
                 internal_port = kafka_security.internal_port(),
                 bootstrap_name = KafkaListenerName::Bootstrap,
                 bootstrap_host = LISTENER_LOCAL_ADDRESS,
-             bootstrap_port = kafka_security.bootstrap_port(),
+                bootstrap_port = kafka_security.bootstrap_port(),
             )
         );
 
@@ -548,7 +548,6 @@ mod tests {
                 internal_protocol = KafkaListenerProtocol::Ssl,
                 bootstrap_name = KafkaListenerName::Bootstrap,
                 bootstrap_protocol = KafkaListenerProtocol::SaslSsl,
-
             )
         );
     }

diff --git a/rust/operator-binary/src/crd/mod.rs b/rust/operator-binary/src/crd/mod.rs
@@ -28,7 +28,7 @@ use stackable_operator::{
     k8s_openapi::{
         api::core::v1::PersistentVolumeClaim, apimachinery::pkg::api::resource::Quantity,
     },
-    kube::{runtime::reflector::ObjectRef, CustomResource, ResourceExt},
+    kube::{CustomResource, ResourceExt, runtime::reflector::ObjectRef},
     product_config_utils::Configuration,
     product_logging::{self, spec::Logging},
     role_utils::{GenericRoleConfig, JavaCommonConfig, Role, RoleGroup, RoleGroupRef},

diff --git a/rust/operator-binary/src/crd/security.rs b/rust/operator-binary/src/crd/security.rs
@@ -7,14 +7,14 @@
 use std::collections::BTreeMap;
 
 use indoc::formatdoc;
-use snafu::{ensure, ResultExt, Snafu};
+use snafu::{ResultExt, Snafu, ensure};
 use stackable_operator::{
     builder::{
         self,
         pod::{
+            PodBuilder,
             container::ContainerBuilder,
             volume::{SecretFormat, SecretOperatorVolumeSourceBuilder, VolumeBuilder},
-            PodBuilder,
         },
     },
     client::Client,
@@ -29,11 +29,12 @@ use stackable_operator::{
 
 use super::listener::node_port_cmd;
 use crate::crd::{
+    KafkaRole, LISTENER_BOOTSTRAP_VOLUME_NAME, LISTENER_BROKER_VOLUME_NAME, SERVER_PROPERTIES_FILE,
+    STACKABLE_CONFIG_DIR, STACKABLE_KERBEROS_KRB5_PATH, STACKABLE_LISTENER_BOOTSTRAP_DIR,
+    STACKABLE_LISTENER_BROKER_DIR, STACKABLE_LOG_DIR,
     authentication::{self, ResolvedAuthenticationClasses},
-    listener::{self, node_address_cmd, KafkaListenerConfig},
-    tls, v1alpha1, KafkaRole, LISTENER_BOOTSTRAP_VOLUME_NAME, LISTENER_BROKER_VOLUME_NAME,
-    SERVER_PROPERTIES_FILE, STACKABLE_CONFIG_DIR, STACKABLE_KERBEROS_KRB5_PATH,
-    STACKABLE_LISTENER_BOOTSTRAP_DIR, STACKABLE_LISTENER_BROKER_DIR, STACKABLE_LOG_DIR,
+    listener::{self, KafkaListenerConfig, node_address_cmd},
+    tls, v1alpha1,
 };
 
 #[derive(Snafu, Debug)]
@@ -66,32 +67,12 @@ pub struct KafkaTlsSecurity {
 }
 
 impl KafkaTlsSecurity {
-    // ports
-    pub const CLIENT_PORT_NAME: &'static str = "kafka";
-    pub const CLIENT_PORT: u16 = 9092;
-    pub const SECURE_CLIENT_PORT_NAME: &'static str = "kafka-tls";
-    pub const SECURE_CLIENT_PORT: u16 = 9093;
+    pub const BOOTSTRAP_PORT: u16 = 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 BOOTSTRAP_PORT: u16 = 9094;
-    pub const SECURE_BOOTSTRAP_PORT: u16 = 9095;
-    // internal
-    pub const INTERNAL_PORT: u16 = 19092;
-    pub const SECURE_INTERNAL_PORT: u16 = 19093;
-    // - TLS global
-    const SSL_STORE_PASSWORD: &'static str = "";
-    // - TLS client
-    const CLIENT_SSL_KEYSTORE_LOCATION: &'static str = "listener.name.client.ssl.keystore.location";
-    const CLIENT_SSL_KEYSTORE_PASSWORD: &'static str = "listener.name.client.ssl.keystore.password";
-    const CLIENT_SSL_KEYSTORE_TYPE: &'static str = "listener.name.client.ssl.keystore.type";
-    const CLIENT_SSL_TRUSTSTORE_LOCATION: &'static str =
-        "listener.name.client.ssl.truststore.location";
-    const CLIENT_SSL_TRUSTSTORE_PASSWORD: &'static str =
-        "listener.name.client.ssl.truststore.password";
-    const CLIENT_SSL_TRUSTSTORE_TYPE: &'static str = "listener.name.client.ssl.truststore.type";
     // - Bootstrapper
     const BOOTSTRAP_SSL_KEYSTORE_LOCATION: &'static str =
         "listener.name.bootstrap.ssl.keystore.location";
@@ -104,6 +85,7 @@ impl KafkaTlsSecurity {
         "listener.name.bootstrap.ssl.truststore.password";
     const BOOTSTRAP_SSL_TRUSTSTORE_TYPE: &'static str =
         "listener.name.bootstrap.ssl.truststore.type";
+    const CLIENT_AUTH_SSL_CLIENT_AUTH: &'static str = "listener.name.client_auth.ssl.client.auth";
     // - TLS client authentication
     const CLIENT_AUTH_SSL_KEYSTORE_LOCATION: &'static str =
         "listener.name.client_auth.ssl.keystore.location";
@@ -117,9 +99,23 @@ impl KafkaTlsSecurity {
         "listener.name.client_auth.ssl.truststore.password";
     const CLIENT_AUTH_SSL_TRUSTSTORE_TYPE: &'static str =
         "listener.name.client_auth.ssl.truststore.type";
-    const CLIENT_AUTH_SSL_CLIENT_AUTH: &'static str = "listener.name.client_auth.ssl.client.auth";
+    pub const CLIENT_PORT: u16 = 9092;
+    // ports
+    pub const CLIENT_PORT_NAME: &'static str = "kafka";
+    // - TLS client
+    const CLIENT_SSL_KEYSTORE_LOCATION: &'static str = "listener.name.client.ssl.keystore.location";
+    const CLIENT_SSL_KEYSTORE_PASSWORD: &'static str = "listener.name.client.ssl.keystore.password";
+    const CLIENT_SSL_KEYSTORE_TYPE: &'static str = "listener.name.client.ssl.keystore.type";
+    const CLIENT_SSL_TRUSTSTORE_LOCATION: &'static str =
+        "listener.name.client.ssl.truststore.location";
+    const CLIENT_SSL_TRUSTSTORE_PASSWORD: &'static str =
+        "listener.name.client.ssl.truststore.password";
+    const CLIENT_SSL_TRUSTSTORE_TYPE: &'static str = "listener.name.client.ssl.truststore.type";
+    // internal
+    pub const INTERNAL_PORT: u16 = 19092;
     // - TLS internal
     const INTER_BROKER_LISTENER_NAME: &'static str = "inter.broker.listener.name";
+    const INTER_SSL_CLIENT_AUTH: &'static str = "listener.name.internal.ssl.client.auth";
     const INTER_SSL_KEYSTORE_LOCATION: &'static str =
         "listener.name.internal.ssl.keystore.location";
     const INTER_SSL_KEYSTORE_PASSWORD: &'static str =
@@ -130,14 +126,19 @@ impl KafkaTlsSecurity {
     const INTER_SSL_TRUSTSTORE_PASSWORD: &'static str =
         "listener.name.internal.ssl.truststore.password";
     const INTER_SSL_TRUSTSTORE_TYPE: &'static str = "listener.name.internal.ssl.truststore.type";
-    const INTER_SSL_CLIENT_AUTH: &'static str = "listener.name.internal.ssl.client.auth";
+    pub const SECURE_BOOTSTRAP_PORT: u16 = 9095;
+    pub const SECURE_CLIENT_PORT: u16 = 9093;
+    pub const SECURE_CLIENT_PORT_NAME: &'static str = "kafka-tls";
+    pub const SECURE_INTERNAL_PORT: u16 = 19093;
+    // - TLS global
+    const SSL_STORE_PASSWORD: &'static str = "";
+    const STACKABLE_TLS_KAFKA_INTERNAL_DIR: &'static str = "/stackable/tls-kafka-internal";
+    const STACKABLE_TLS_KAFKA_INTERNAL_VOLUME_NAME: &'static str = "tls-kafka-internal";
+    const STACKABLE_TLS_KAFKA_SERVER_DIR: &'static str = "/stackable/tls-kafka-server";
+    const STACKABLE_TLS_KAFKA_SERVER_VOLUME_NAME: &'static str = "tls-kafka-server";
     // directories
     const STACKABLE_TLS_KCAT_DIR: &'static str = "/stackable/tls-kcat";
     const STACKABLE_TLS_KCAT_VOLUME_NAME: &'static str = "tls-kcat";
-    const STACKABLE_TLS_KAFKA_SERVER_DIR: &'static str = "/stackable/tls-kafka-server";
-    const STACKABLE_TLS_KAFKA_SERVER_VOLUME_NAME: &'static str = "tls-kafka-server";
-    const STACKABLE_TLS_KAFKA_INTERNAL_DIR: &'static str = "/stackable/tls-kafka-internal";
-    const STACKABLE_TLS_KAFKA_INTERNAL_VOLUME_NAME: &'static str = "tls-kafka-internal";
 
     #[cfg(test)]
     pub fn new(
@@ -685,7 +686,9 @@ impl KafkaTlsSecurity {
             "-X".to_string(),
             format!("sasl.kerberos.service.name={service_name}"),
             "-X".to_string(),
-            format!("sasl.kerberos.principal={service_name}/$POD_BROKER_LISTENER_ADDRESS@$KERBEROS_REALM"),
+            format!(
+                "sasl.kerberos.principal={service_name}/$POD_BROKER_LISTENER_ADDRESS@$KERBEROS_REALM"
+            ),
         ]
     }
 }
diff --git a/rust/operator-binary/src/discovery.rs b/rust/operator-binary/src/discovery.rs
@@ -5,11 +5,11 @@ use stackable_operator::{
     builder::{configmap::ConfigMapBuilder, meta::ObjectMetaBuilder},
     commons::{listener::Listener, product_image_selection::ResolvedProductImage},
     k8s_openapi::api::core::v1::{ConfigMap, Service},
-    kube::{runtime::reflector::ObjectRef, Resource, ResourceExt},
+    kube::{Resource, ResourceExt, runtime::reflector::ObjectRef},
 };
 
 use crate::{
-    crd::{security::KafkaTlsSecurity, v1alpha1, KafkaRole},
+    crd::{KafkaRole, security::KafkaTlsSecurity, v1alpha1},
     kafka_controller::KAFKA_CONTROLLER_NAME,
     utils::build_recommended_labels,
 };