feat: kerberized Kraft controllers

CHANGELOG.md

@@ -8,6 +8,8 @@ All notable changes to this project will be documented in this file.
 
 - Support floating tags for product images via the new `spec.image.stackableVersionPolicy` field
   ([#1021]).
+- Support Kerberos (GSSAPI) authentication on KRaft controllers, covering both
+  broker-to-controller and controller-to-controller (Raft) traffic ([#1024]).
 - Support for Kafka 4.3.1 ([#1022]).
 
 ### Changed
@@ -22,9 +24,6 @@ All notable changes to this project will be documented in this file.
   that adds the new controller to the voter list.
   On termination, a new `preStop` hook on the controller container (`kafka`) removes the pod from
   the voter list before shutdown.
-  The property `controller.quorum.bootstrap.servers` now contains the headless service names
-  of all controller role groups instead of individual peer host names. This prevents the
-  restart controller from restarting all pods in the quorum when a new one is added/deleted.
   The controller `StatefulSet` is now scaled using `OrderedReady` instead of the `Parallel` strategy
   to ensure only one voter is added/removed at a time and thus keep the quorum healthy ([#1010]).
 - Internal operator refactoring: introduce a build() step in the reconciler that
@@ -79,6 +78,7 @@ All notable changes to this project will be documented in this file.
 [#1017]: https://github.com/stackabletech/kafka-operator/pull/1017
 [#1021]: https://github.com/stackabletech/kafka-operator/pull/1021
 [#1022]: https://github.com/stackabletech/kafka-operator/pull/1022
+[#1024]: https://github.com/stackabletech/kafka-operator/pull/1024
 
 ## [26.7.0] - 2026-07-21
 

docs/modules/kafka/pages/usage-guide/kraft-controller.adoc

@@ -92,13 +92,34 @@ KRaft mode requires major configuration changes compared to ZooKeeper:
   with `kafka-storage.sh format --standalone`, bootstrapping a single-node quorum by itself.
   Every other controller formats with `--no-initial-controllers` and joins purely through the sidecar's `add-controller` call.
   Brokers always format with `--no-initial-controllers` too; they are never voters.
-* `controller.quorum.bootstrap.servers` points at each controller role group's own headless Service DNS name, not
-  individual pod addresses.
+* `controller.quorum.bootstrap.servers` lists the individual pod FQDNs of all controllers.
+  This is required by Kerberos. A GSSAPI client derives the service principal from the hostname it dials, and the
+  CONTROLLER listener can only offer the pod's own principal. But it means the property changes whenever a
+  controller role group is scaled, so the controller pods are rolled by the restart controller on scale.
+
+== Kerberos
+
+Apache Kafka KRaft controllers support Kerberos (GSSAPI) authentication.
+Reference an `AuthenticationClass` with the Kerberos provider as described in
+xref:usage-guide/security.adoc[Security].
+
+When Kerberos is enabled:
+
+* The `CONTROLLER` listener uses `SASL_SSL` instead of `SSL`, with `GSSAPI` as the mechanism
+  (`sasl.mechanism.controller.protocol`). This covers both broker-to-controller traffic and
+  the controller-to-controller (Raft) traffic between quorum members.
+* Controller keytabs are *pod-scoped*, unlike broker keytabs, which are scoped to their
+  listener volumes. Controllers are reachable only under their own StatefulSet pod DNS name,
+  which is therefore the principal in their keytab.
+* Each pod's `jaas.properties` gains a `controller.KafkaServer` login context. Unlike the
+  broker-side contexts it does not set `isInitiator=false`, because a controller must be able
+  to initiate GSSAPI connections to its peers, not only accept them.
+* The `quorum-manager` sidecar and the `preStop` hook authenticate to the CONTROLLER listener
+  as the controller's own pod principal.
 
 == Known Issues
 
 * Automatic migration from Apache ZooKeeper to KRaft is not supported.
-* Kerberos is currently not supported for KRaft in all versions.
 * The quorum is created once by the controller with the lowest `node.id` using `--standalone`. If this controller
   loses it's PVC, a new conflicting quorum is created on restart.
 * A Controller that loses its persistent volume is not re-admitted to the voter set automatically, because it

docs/modules/kafka/pages/usage-guide/security.adoc

@@ -162,13 +162,55 @@ NOTE: When Kerberos is enabled it is also required to enable TLS for maximum sec
 
 ==== Clients
 
-In order to keep client configuration as uncluttered as possible, each kerberized Kafka broker has two principals: one for the broker itself and one for the bootstrap service.
-The client can connect to the bootstrap service, which returns the broker quorum for use in subsequent operations.
-This is transparent as each connection dynamically uses the relevant principal (broker or bootstrap).
-In order for this to work, it is necessary for kerberized clusters to define an extra Kafka listener for the bootstrap with a corresponding service (and port).
-The bootstrap address is written to the discovery ConfigMap, using the Stackable bootstrap listener with the port being 9095 (secure) for kerberized clusters, and 9092 (non-secure) or 9093 (secure) for non-kerberized ones.
+===== Why each broker has two principals
 
-NOTE: Port 9094 is reserved for non-secure kerberized connections which is not currently implemented.
+A GSSAPI client derives the service principal it asks the KDC for from the *hostname it connects to*.
+Connecting to a Kafka cluster takes two hops over two different addresses: a client first contacts a bootstrap address, receives the cluster metadata, and then connects to individual brokers directly.
+Each of those addresses therefore needs its own principal.
+
+Every broker consequently gets two principals, `kafka/<bootstrap-address>` and `kafka/<broker-address>`, both provisioned into its keytab by the Secret Operator.
+The broker's JAAS configuration declares both, as the `bootstrap.KafkaServer` and `client.KafkaServer` login contexts.
+No client-side configuration is needed to switch between them: whichever address a client dials, the broker already holds the matching principal.
+
+To make this work, a kerberized cluster exposes an additional Kafka listener, container port and `Listener` port for the bootstrap address.
+These exist only when Kerberos is enabled; without it, clients reach brokers over the client listener alone.
+
+===== Bootstrap address and ports
+
+The bootstrap address is published in the xref:reference/discovery.adoc[discovery ConfigMap] under the `KAFKA` key, read from the ingress addresses of the Stackable bootstrap `Listener`.
+The port depends on whether Kerberos and TLS are enabled:
+
+[cols="1,1,1"]
+|===
+| Cluster | Port name | Port
+
+| Kerberos (always TLS)
+| `bootstrap`
+| 9095
+
+| TLS, no Kerberos
+| `kafka-tls`
+| 9093
+
+| No TLS, no Kerberos
+| `kafka`
+| 9092
+|===
+
+NOTE: Kerberos requires TLS, so a kerberized cluster always uses port 9095 for bootstrapping.
+
+===== Client configuration
+
+The discovery ConfigMap's `client.properties` carries the properties needed to *reach* the cluster:
+
+- `security.protocol`
+- `sasl.mechanism`
+- `sasl.kerberos.service.name`
+- and the truststore settings
+
+But the operator adds *no login configuration* to `client.properties` because supplying credentials is the client's responsibility.
+The operator cannot do it: these clients run outside the Kafka Pods, so they have neither the Pods' keytabs nor their principals.
+Add your own `sasl.jaas.config`, or point the JVM at a JAAS file with `java.security.auth.login.config`, naming the principal and keytab the client should authenticate with.
 
 == [[authorization]]Authorization
 

docs/modules/kafka/partials/supported-versions.adoc

@@ -6,6 +6,6 @@
 * 4.2.1
 * 3.9.2 (LTS)
 
-Starting with Apache Kafka >=4.x, the quorum manager has been changed from Apache Zookeeper to the built in KRaft.
+Starting with Apache Kafka 4, the quorum manager has been changed from Apache Zookeeper to the built-in KRaft.
 
 If you are running an older Kafka version and plan to upgrade, head on to the xref:kafka:usage-guide/kraft-controller.adoc[Kraft migration guide] for details on how to do so.

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

@@ -43,6 +43,14 @@ const DERIVE_POD_INDEX: &str = r#"POD_INDEX=$(echo "$POD_NAME" | grep -oE '[0-9]
 
 const EXPORT_REPLICA_ID: &str = "export REPLICA_ID=$((POD_INDEX + NODE_ID_OFFSET))";
 
+/// Shell snippet that safely exports `$KERBEROS_REALM`, extracted from the `default_realm`
+/// line of the mounted krb5.conf.
+pub fn set_kerberos_realm_env_command() -> String {
+    format!(
+        "KERBEROS_REALM=$(grep -oP 'default_realm = \\K.*' {STACKABLE_KERBEROS_KRB5_PATH} 2>/dev/null) && export KERBEROS_REALM || true"
+    )
+}
+
 /// Returns the commands to start the main Kafka container
 pub fn broker_kafka_container_commands(
     kraft_mode: bool,
@@ -65,8 +73,8 @@ pub fn broker_kafka_container_commands(
         remove_vector_shutdown_file_command = remove_vector_shutdown_file_command(STACKABLE_LOG_DIR),
         create_vector_shutdown_file_command = create_vector_shutdown_file_command(STACKABLE_LOG_DIR),
         set_realm_env = match kafka_security.has_kerberos_enabled() {
-            true => format!("export KERBEROS_REALM=$(grep -oP 'default_realm = \\K.*' {STACKABLE_KERBEROS_KRB5_PATH})"),
-            false => "".to_string(),
+            true => set_kerberos_realm_env_command(),
+            false => String::new(),
         },
         import_opa_tls_cert = copy_opa_tls_cert_command(kafka_security),
         broker_start_command = broker_start_command(kraft_mode),
@@ -142,13 +150,15 @@ fn controller_quorum_format_flag(controller_descriptors: &[KafkaPodDescriptor])
 }
 
 pub fn controller_kafka_container_command(
+    kafka_security: &ValidatedKafkaSecurity,
     controller_descriptors: Vec<KafkaPodDescriptor>,
 ) -> String {
     formatdoc! {"
         {COMMON_BASH_TRAP_FUNCTIONS}
         {remove_vector_shutdown_file_command}
         prepare_signal_handlers
         containerdebug --output={STACKABLE_LOG_DIR}/containerdebug-state.json --loop &
+        {set_realm_env}
 
         {derive_pod_index}
         {export_replica_id}
@@ -157,6 +167,12 @@ pub fn controller_kafka_container_command(
 
         config-utils template /tmp/{properties_file}
 
+        cp {config_dir}/{jaas_file} /tmp/{jaas_file}
+        config-utils template /tmp/{jaas_file}
+
+        cp {admin_client_source} {admin_client_config}
+        config-utils template {admin_client_config}
+
         {quorum_format_flag}
         bin/kafka-storage.sh format --cluster-id \"$KAFKA_CLUSTER_ID\" --config /tmp/{properties_file} --ignore-formatted \"$FORMAT_QUORUM_FLAG\"
         bin/kafka-server-start.sh /tmp/{properties_file} &
@@ -165,18 +181,30 @@ pub fn controller_kafka_container_command(
         {create_vector_shutdown_file_command}
         ",
         remove_vector_shutdown_file_command = remove_vector_shutdown_file_command(STACKABLE_LOG_DIR),
+        // Mirrors `broker_kafka_container_commands`: empty when Kerberos is disabled.
+        set_realm_env = match kafka_security.has_kerberos_enabled() {
+            true => set_kerberos_realm_env_command(),
+            false => String::new(),
+        },
         derive_pod_index = DERIVE_POD_INDEX,
         export_replica_id = EXPORT_REPLICA_ID,
         config_dir = STACKABLE_CONFIG_DIR,
         properties_file = ConfigFileName::ControllerProperties,
+        jaas_file = ConfigFileName::Jaas,
+        admin_client_source = ADMIN_CLIENT_PROPERTIES_SOURCE_PATH,
+        admin_client_config = ADMIN_CLIENT_PROPERTIES_PATH,
         quorum_format_flag = controller_quorum_format_flag(&controller_descriptors),
         create_vector_shutdown_file_command = create_vector_shutdown_file_command(STACKABLE_LOG_DIR)
     }
 }
 
 const KAFKA_METADATA_QUORUM_BINARY: &str = "/stackable/kafka/bin/kafka-metadata-quorum.sh";
 
-const ADMIN_CLIENT_PROPERTIES_PATH: &str = "/stackable/config/admin-client.properties";
+/// The rendered admin-client config. The raw ConfigMap file is copied here and passed through
+/// `config-utils template` first, because under Kerberos its `sasl.jaas.config` carries
+/// `${env:...}` placeholders (see `controller_admin_client_properties`).
+const ADMIN_CLIENT_PROPERTIES_PATH: &str = "/tmp/admin-client.properties";
+const ADMIN_CLIENT_PROPERTIES_SOURCE_PATH: &str = "/stackable/config/admin-client.properties";
 
 /// The merged config used only for `add-controller` (self-registration).
 ///
@@ -200,11 +228,6 @@ const CLI_CALL_KILL_AFTER_SECONDS: u32 = 5;
 /// Shell snippet setting `$BOOTSTRAP_SERVERS` by extracting
 /// `controller.quorum.bootstrap.servers` from the static, un-rendered `controller.properties`
 /// ConfigMap file.
-///
-/// Reading this at runtime, rather than baking the peer list into this script as a Rust
-/// literal, keeps both sidecar scripts' content — and therefore the controller pod
-/// template — identical across changes to an existing controller role group's *replica
-/// count*.
 fn extract_bootstrap_servers_command() -> String {
     format!(
         r#"BOOTSTRAP_SERVERS=$(grep '^controller.quorum.bootstrap.servers=' {config_dir}/{controller_properties_file} | cut -d= -f2- | sed 's/\\:/:/g')"#,
@@ -230,6 +253,10 @@ const CONTROLLER_QUORUM_MANAGER_LOOP_SCRIPT: &str =
 /// The sidecar's main-loop command: while this controller's local Raft state is `observer`,
 /// admit it into the quorum's voter set once that is safe.
 pub fn quorum_manager_container_command() -> String {
+    // The sidecar is a separate container and inherits nothing from the kafka container's
+    // startup, so it derives the realm itself. Harmless when krb5.conf is absent: only the
+    // Kerberos case has a `${env:KERBEROS_REALM}` placeholder for `config-utils` to resolve.
+    let set_realm_env = set_kerberos_realm_env_command();
     format!(
         r#"
         set -uo pipefail
@@ -238,9 +265,12 @@ pub fn quorum_manager_container_command() -> String {
         [ -n "$POD_INDEX" ] || exit 0
         {export_replica_id}
         {extract_bootstrap_servers}
+        {set_realm_env}
 
         if cp {config_dir}/{controller_properties_file} /tmp/{controller_properties_file} \
           && config-utils template /tmp/{controller_properties_file} \
+          && cp {admin_client_source} {admin_client_config} \
+          && config-utils template {admin_client_config} \
           && cat /tmp/{controller_properties_file} {admin_client_config} > {add_controller_config}; then
           QUORUM_CLI={binary}
           ADMIN_CLIENT_CONFIG={admin_client_config}
@@ -266,8 +296,10 @@ pub fn quorum_manager_container_command() -> String {
         derive_pod_index = DERIVE_POD_INDEX,
         export_replica_id = EXPORT_REPLICA_ID,
         extract_bootstrap_servers = extract_bootstrap_servers_command(),
+        set_realm_env = set_realm_env,
         config_dir = STACKABLE_CONFIG_DIR,
         controller_properties_file = ConfigFileName::ControllerProperties,
+        admin_client_source = ADMIN_CLIENT_PROPERTIES_SOURCE_PATH,
         admin_client_config = ADMIN_CLIENT_PROPERTIES_PATH,
         add_controller_config = ADD_CONTROLLER_PROPERTIES_PATH,
         cli_timeout = CLI_CALL_TIMEOUT_SECONDS,
@@ -378,6 +410,65 @@ mod tests {
     use indoc::indoc;
 
     use super::*;
+    use crate::controller::build::security::tests::{kerberos, plaintext};
+
+    #[test]
+    fn controller_command_exports_the_kerberos_realm_when_enabled() {
+        let command = controller_kafka_container_command(&kerberos(), vec![]);
+        assert!(command.contains(&set_kerberos_realm_env_command()));
+    }
+
+    #[test]
+    fn controller_command_does_not_export_a_realm_without_kerberos() {
+        let command = controller_kafka_container_command(&plaintext(), vec![]);
+        assert!(!command.contains("KERBEROS_REALM"));
+    }
+
+    #[test]
+    fn controller_command_always_templates_the_jaas_file() {
+        // `jaas.properties` is always present in the ConfigMap (empty when Kerberos is off),
+        // so the copy is unconditional, matching `broker_start_command`.
+        for security in [kerberos(), plaintext()] {
+            let command = controller_kafka_container_command(&security, vec![]);
+            assert!(command.contains("cp /stackable/config/jaas.properties /tmp/jaas.properties"));
+            assert!(command.contains("config-utils template /tmp/jaas.properties"));
+        }
+    }
+
+    #[test]
+    fn quorum_manager_templates_the_admin_client_config() {
+        let command = quorum_manager_container_command();
+        assert!(
+            command.contains(
+                "cp /stackable/config/admin-client.properties /tmp/admin-client.properties"
+            )
+        );
+        assert!(command.contains("config-utils template /tmp/admin-client.properties"));
+        // It must connect with the *rendered* copy, not the raw ConfigMap file, or the
+        // `${env:...}` placeholders in `sasl.jaas.config` reach the JAAS parser verbatim.
+        assert!(command.contains("ADMIN_CLIENT_CONFIG=/tmp/admin-client.properties"));
+        assert!(!command.contains("ADMIN_CLIENT_CONFIG=/stackable/config/admin-client.properties"));
+    }
+
+    #[test]
+    fn quorum_manager_exports_the_kerberos_realm() {
+        // The sidecar is a separate container: it inherits nothing from the kafka container's
+        // startup, so it must derive $KERBEROS_REALM itself for `config-utils template` to
+        // resolve the principal.
+        let command = quorum_manager_container_command();
+        assert!(command.contains("KERBEROS_REALM"));
+    }
+
+    #[test]
+    fn controller_command_templates_the_admin_client_config_for_pre_stop() {
+        let command = controller_kafka_container_command(&kerberos(), vec![]);
+        assert!(
+            command.contains(
+                "cp /stackable/config/admin-client.properties /tmp/admin-client.properties"
+            )
+        );
+        assert!(command.contains("config-utils template /tmp/admin-client.properties"));
+    }
 
     #[test]
     fn quorum_manager_container_command_targets_the_bootstrap_servers_not_localhost() {
@@ -446,11 +537,11 @@ mod tests {
         // `listeners`) via the same REPLICA_ID derivation used by the `kafka` container.
         assert!(command.contains("export REPLICA_ID=$((POD_INDEX + NODE_ID_OFFSET))"));
         assert!(command.contains("config-utils template /tmp/controller.properties"));
-        // Merges it with the plain admin-client config (carries `security.protocol`/`ssl.*`),
-        // controller.properties first so the client TLS config in admin-client.properties
-        // wins on any key collision (see `ADD_CONTROLLER_PROPERTIES_PATH`'s doc comment).
+        // Merges it with the *rendered* admin-client config (carries `security.protocol`,
+        // `ssl.*` and, under Kerberos, `sasl.jaas.config`), controller.properties first so
+        // the client config wins on any key collision (see `ADD_CONTROLLER_PROPERTIES_PATH`).
         assert!(command.contains(
-            "cat /tmp/controller.properties /stackable/config/admin-client.properties > /tmp/add-controller.properties"
+            "cat /tmp/controller.properties /tmp/admin-client.properties > /tmp/add-controller.properties"
         ));
         // The merged file is what `add-controller` — and only `add-controller` — connects
         // with; read-only `describe` calls keep using the plain admin-client config.
@@ -1119,7 +1210,7 @@ mod tests {
             pod_descriptor(KafkaRole::Controller, 1, 6),
             pod_descriptor(KafkaRole::Controller, 2, 7),
         ];
-        let command = controller_kafka_container_command(descriptors);
+        let command = controller_kafka_container_command(&plaintext(), descriptors);
 
         assert!(command.contains(r#"if [ "$REPLICA_ID" = "5" ]; then"#));
         assert!(command.contains("FORMAT_QUORUM_FLAG=--standalone"));