From c4a1fc3dc47a1ed45948036c5123405e064802d9 Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Fri, 7 Aug 2026 15:13:39 +0200 Subject: [PATCH 01/14] docs: add KRaft controller Kerberos implementation plan Co-Authored-By: Claude Sonnet 5 --- .../2026-08-07-kraft-kerberos-support.md | 971 ++++++++++++++++++ 1 file changed, 971 insertions(+) create mode 100644 docs/superpowers/plans/2026-08-07-kraft-kerberos-support.md diff --git a/docs/superpowers/plans/2026-08-07-kraft-kerberos-support.md b/docs/superpowers/plans/2026-08-07-kraft-kerberos-support.md new file mode 100644 index 00000000..f6946e30 --- /dev/null +++ b/docs/superpowers/plans/2026-08-07-kraft-kerberos-support.md @@ -0,0 +1,971 @@ +# KRaft Controller Kerberos Support Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Make Kerberos (GSSAPI) authentication work on the `CONTROLLER` listener, so KRaft clusters (`spec.controllers` / `metadataManager: kraft`) can be secured with Kerberos exactly like ZooKeeper-based clusters already are, and drop "Kerberos is not supported for KRaft" from the docs. + +**Architecture:** The `CONTROLLER` listener is a normal Kafka listener governed by `listener.security.protocol.map`; Kafka's own docs document `CONTROLLER:SASL_SSL` as a supported combination (see Research Findings below). Closing the gap means: (1) letting the `CONTROLLER` listener protocol switch to `SASL_SSL` when Kerberos is enabled, (2) generating the extra Kafka/JAAS properties Kafka needs for controller-listener SASL, and (3) actually mounting a Kerberos keytab + JAAS file into KRaft controller pods, which today get none of that (Kerberos wiring in this operator is currently broker-only, end to end). + +**Tech Stack:** Rust (`operator-binary`), existing `stackable-operator` secret-operator volume builder, Kafka `controller.properties`/`broker.properties`, static JAAS file convention already used by this operator (`.KafkaServer` sections in `jaas.properties`). + +## Research Findings (why this is answerable positively) + +1. Apache Kafka's official listener docs (`kafka.apache.org/41/security/listener-configuration/`) explicitly document `listener.security.protocol.map=BROKER:SASL_SSL,CONTROLLER:SASL_SSL` as a valid KRaft configuration. There is **no Kafka-level restriction** on using SASL/GSSAPI on the controller listener. +2. The real, documented KRaft/SASL limitation is specific to **SASL/SCRAM**, not GSSAPI: [KAFKA-15513](https://issues.apache.org/jira/browse/KAFKA-15513) — SCRAM credentials live in the `__cluster_metadata` log, which isn't readable yet while controllers are still forming quorum, so SCRAM control-plane auth is genuinely broken. GSSAPI/Kerberos has no such bootstrap problem: it authenticates against an external KDC, not against Kafka-internal credential storage, so it doesn't hit that chicken-and-egg issue. +3. Conclusion: **Kerberos on KRaft controllers is not blocked by Kafka — it's an implementation gap in this operator.** The `docs/modules/kafka/usage-guide/kraft-controller.adoc` "Known Issues" bullet and the `KafkaListenerName::Controller` doc comment ("this listener does not support SSL_SASL") are both operator-authored claims, not upstream Kafka facts, and both are simply wrong once this plan is implemented. +4. Concretely, today's gap (verified by reading the code, not guessing): + - `get_kafka_listener_config()` (`controller/build/properties/listener.rs:111-112`) hardcodes `KafkaListenerName::Controller → KafkaListenerProtocol::Ssl`, never checking `has_kerberos_enabled()`. + - Kafka has a distinct config key `sasl.mechanism.controller.protocol` (separate from `sasl.mechanism.inter.broker.protocol`) that this operator never sets. + - `add_kerberos_pod_config()` (mounts the keytab/krb5.conf volume, sets `KRB5_CONFIG`/`KAFKA_OPTS`) is only ever called from `build_broker_rolegroup_statefulset` — never from `build_controller_rolegroup_statefulset`. Controller pods get **no keytab at all** today. + - `controller_kafka_container_command()` never exports `KERBEROS_REALM` and never copies/templates `jaas.properties` into `/tmp`, unlike the broker startup command. + - `jaas_config_file()` only ever emits `bootstrap.KafkaServer` and `client.KafkaServer` JAAS sections — there is no `controller.KafkaServer` section for the CONTROLLER listener. + - Controller pods have no listener-operator `Listener` volume (only brokers do — confirmed: "Only broker role groups get a bootstrap Listener", `controller/build/mod.rs:130`); their Kerberos keytab (once added) must be a **pod-scoped** secret-operator volume (like the controller's existing internal-TLS cert, `add_controller_volume_and_volume_mounts`, `controller/build/security.rs:303-337`), not a listener-volume-scoped one. + - `controller_config_settings()` (`controller/build/security.rs:503-548`) already sets `sasl.enabled.mechanisms`/`sasl.kerberos.service.name`/`sasl.mechanism.inter.broker.protocol` for controllers when Kerberos is enabled — this part is already correct and should be left as-is except for the addition in Task 3. + +## Global Constraints + +- Match existing Rust/Stackable conventions in this repo (snafu errors, `BTreeMap` config builders, existing test style using `ValidatedKafkaSecurity::new(...)` fixtures) — see `stackable-development-plugin:stackable-rust-style`. +- No new external dependencies. +- Every behavioural change must be covered by a `#[cfg(test)]` unit test in the same module, following the existing table of fixtures (`plaintext()`, `kerberos()`, `internal_tls()`, etc. in `controller/build/security.rs`, and `test_get_kafka_kerberos_listeners_config` in `controller/build/properties/listener.rs`). +- Do not touch ZooKeeper-mode Kerberos behavior — every change must be gated so plaintext/TLS/non-Kraft/non-Kerberos configurations produce byte-identical output to before. +- Docs (`docs/modules/kafka/...`) and `CHANGELOG.md` must be updated in the same PR that lands the feature, not deferred. + +--- + +### Task 1: Correct the `CONTROLLER` listener's Kerberos support to `SASL_SSL` + +**Files:** +- Modify: `rust/operator-binary/src/crd/listener.rs:58-69` (doc comment) +- Modify: `rust/operator-binary/src/controller/build/properties/listener.rs:110-112` +- Modify: `rust/operator-binary/src/controller/build/properties/listener.rs:484-497` (existing test, currently asserts the old/wrong behavior) +- Test: same file, new assertion added to `test_get_kafka_kerberos_listeners_config` + +**Interfaces:** +- Consumes: `ValidatedKafkaSecurity::has_kerberos_enabled()` (already exists, `controller/security.rs`). +- Produces: `KafkaListenerConfig.listener_security_protocol_map` now maps `Controller → SaslSsl` whenever Kerberos is enabled (consumed by later tasks and by `controller_config_settings()`/`broker_config_settings()`, which already read `security.has_kerberos_enabled()` independently). + +- [ ] **Step 1: Update the doc comment on `KafkaListenerName::Controller`** + +In `rust/operator-binary/src/crd/listener.rs`, replace: + +```rust + /// This listener is defined when Kraft mode is enabled. + /// It is responsible for broker/controller as well as controller/controller communications + /// and therefore it is present on *both* brokers and controller properties files. + /// The only protocol used is SSL. + /// The advertised host names are FQDN pod names of the controllers. + /// + /// Notes: + /// + /// - there is no listener for client/controller communication + /// - this listener does not support SSL_SASL. + #[strum(serialize = "CONTROLLER")] + Controller, +``` + +with: + +```rust + /// This listener is defined when Kraft mode is enabled. + /// It is responsible for broker/controller as well as controller/controller communications + /// and therefore it is present on *both* brokers and controller properties files. + /// The protocol used is SSL, or SASL_SSL when Kerberos is enabled. + /// The advertised host names are FQDN pod names of the controllers. + /// + /// Note: there is no listener for client/controller communication. + #[strum(serialize = "CONTROLLER")] + Controller, +``` + +- [ ] **Step 2: Write the failing test** + +In `rust/operator-binary/src/controller/build/properties/listener.rs`, change the `controller_protocol` expectation inside `test_get_kafka_kerberos_listeners_config` (currently at line ~495) from: + +```rust + controller_name = KafkaListenerName::Controller, + controller_protocol = KafkaListenerProtocol::Ssl, +``` + +to: + +```rust + controller_name = KafkaListenerName::Controller, + controller_protocol = KafkaListenerProtocol::SaslSsl, +``` + +- [ ] **Step 3: Run test to verify it fails** + +Run: `cd rust && cargo test -p stackable-kafka-operator-binary controller::build::properties::listener::tests::test_get_kafka_kerberos_listeners_config -- --exact` +Expected: FAIL — assertion left/right mismatch (`Ssl` vs `SaslSsl`). + +- [ ] **Step 4: Implement the minimal fix** + +In `rust/operator-binary/src/controller/build/properties/listener.rs`, replace: + +```rust + listener_security_protocol_map.insert(KafkaListenerName::Internal, KafkaListenerProtocol::Ssl); + listener_security_protocol_map + .insert(KafkaListenerName::Controller, KafkaListenerProtocol::Ssl); +``` + +with: + +```rust + listener_security_protocol_map.insert(KafkaListenerName::Internal, KafkaListenerProtocol::Ssl); + listener_security_protocol_map.insert( + KafkaListenerName::Controller, + if kafka_security.has_kerberos_enabled() { + KafkaListenerProtocol::SaslSsl + } else { + KafkaListenerProtocol::Ssl + }, + ); +``` + +- [ ] **Step 5: Run test to verify it passes** + +Run: `cd rust && cargo test -p stackable-kafka-operator-binary controller::build::properties::listener::tests:: -- --exact` (runs both listener tests in the module) +Expected: PASS. + +- [ ] **Step 6: Commit** + +```bash +git add rust/operator-binary/src/crd/listener.rs rust/operator-binary/src/controller/build/properties/listener.rs +git commit -m "feat: allow SASL_SSL on the KRaft CONTROLLER listener when Kerberos is enabled" +``` + +--- + +### Task 2: Set `sasl.mechanism.controller.protocol` on brokers and controllers + +Kafka distinguishes `sasl.mechanism.inter.broker.protocol` (used for the `INTERNAL`/inter-broker listener) from `sasl.mechanism.controller.protocol` (used specifically for the `CONTROLLER` listener). The operator currently never sets the latter; add it wherever the former is already set for Kerberos. + +**Files:** +- Modify: `rust/operator-binary/src/controller/build/security.rs:46-51` (constants), `:432-452` (`broker_config_settings`), `:530-545` (`controller_config_settings`) +- Test: same file, extend `broker_config_kerberos_adds_sasl_and_bootstrap_stores` and `controller_config_kerberos_adds_sasl` (existing tests around lines 921 and 980) + +**Interfaces:** +- Consumes: `ValidatedKafkaSecurity::has_kerberos_enabled()` (unchanged). +- Produces: `broker_config_settings()` and `controller_config_settings()` both now include `sasl.mechanism.controller.protocol=GSSAPI` in their returned `BTreeMap` whenever Kerberos is enabled — no other caller needs to change. + +- [ ] **Step 1: Write the failing tests** + +In `rust/operator-binary/src/controller/build/security.rs`, extend the existing test: + +```rust + #[test] + fn broker_config_kerberos_adds_sasl_and_bootstrap_stores() { + let config = broker_config_settings(&kerberos()); + assert_eq!( + config.get("sasl.enabled.mechanisms"), + Some(&"GSSAPI".to_string()) + ); + assert_eq!( + config.get("sasl.kerberos.service.name"), + Some(&"kafka".to_string()) + ); + assert_eq!( + config.get("sasl.mechanism.inter.broker.protocol"), + Some(&"GSSAPI".to_string()) + ); + assert_eq!( + config.get("sasl.mechanism.controller.protocol"), + Some(&"GSSAPI".to_string()) + ); + assert!(config.contains_key("listener.name.bootstrap.ssl.keystore.location")); + } +``` + +and: + +```rust + #[test] + fn controller_config_kerberos_adds_sasl() { + let config = controller_config_settings(&kerberos()); + assert_eq!( + config.get("sasl.enabled.mechanisms"), + Some(&"GSSAPI".to_string()) + ); + assert_eq!( + config.get("sasl.kerberos.service.name"), + Some(&"kafka".to_string()) + ); + assert_eq!( + config.get("sasl.mechanism.controller.protocol"), + Some(&"GSSAPI".to_string()) + ); + } +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `cd rust && cargo test -p stackable-kafka-operator-binary controller::build::security::tests::broker_config_kerberos_adds_sasl_and_bootstrap_stores controller::build::security::tests::controller_config_kerberos_adds_sasl` +Expected: FAIL — `sasl.mechanism.controller.protocol` key missing (`None` vs `Some("GSSAPI")`). + +- [ ] **Step 3: Implement** + +Add the constant next to the others in `rust/operator-binary/src/controller/build/security.rs`: + +```rust +const PROPERTY_SASL_CONTROLLER_MECHANISM: &str = "sasl.mechanism.controller.protocol"; +``` + +In `broker_config_settings()`, inside the existing `if security.has_kerberos_enabled() { ... }` block (around line 439-450), add: + +```rust + config.insert( + PROPERTY_SASL_CONTROLLER_MECHANISM.to_string(), + SASL_MECHANISM_GSSAPI.to_string(), + ); +``` + +In `controller_config_settings()`, inside its `if security.has_kerberos_enabled() { ... }` block (around line 532-543), add the same insert. + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `cd rust && cargo test -p stackable-kafka-operator-binary controller::build::security::tests::` +Expected: PASS for all tests in the module (nothing else should have changed). + +- [ ] **Step 5: Commit** + +```bash +git add rust/operator-binary/src/controller/build/security.rs +git commit -m "feat: set sasl.mechanism.controller.protocol for Kerberos-enabled clusters" +``` + +--- + +### Task 3: Make `add_kerberos_pod_config` support pod-scoped volumes (for controllers, no kcat container) + +Today `add_kerberos_pod_config` (`controller/build/kerberos.rs`) always scopes the keytab to the two *listener* volumes (`LISTENER_BROKER_VOLUME_NAME`, `LISTENER_BOOTSTRAP_VOLUME_NAME`) and always mounts into a kcat-prober container. Controller pods have neither of those — they need a **pod-scoped** keytab (same secret-operator pattern already used for the controller's internal TLS cert, `add_controller_volume_and_volume_mounts`) and have no kcat-prober container at all. + +**Files:** +- Modify: `rust/operator-binary/src/controller/build/kerberos.rs` +- Test: same file — this module currently has no `#[cfg(test)]` block; add one. + +**Interfaces:** +- Consumes: `KafkaRole` (already a parameter), `ValidatedKafkaSecurity::kerberos_secret_class()` (already exists). +- Produces: `add_kerberos_pod_config(kafka_security, role, cb_kcat_prober: Option<&mut ContainerBuilder>, cb_kafka: &mut ContainerBuilder, pb: &mut PodBuilder) -> Result<(), Error>` — signature changes from `cb_kcat_prober: &mut ContainerBuilder` to `Option<&mut ContainerBuilder>`. Task 4 and the existing broker call site both depend on this new signature. + +- [ ] **Step 1: Write the failing test** + +Add to `rust/operator-binary/src/controller/build/kerberos.rs`: + +```rust +#[cfg(test)] +mod tests { + use stackable_operator::{ + builder::{meta::ObjectMetaBuilder, pod::container::ContainerBuilder}, + crd::authentication::{core, kerberos}, + }; + + use super::*; + use crate::crd::authentication::ResolvedAuthenticationClasses; + + fn kerberos_security() -> ValidatedKafkaSecurity { + ValidatedKafkaSecurity::new( + ResolvedAuthenticationClasses::new(vec![core::v1alpha1::AuthenticationClass { + metadata: ObjectMetaBuilder::new().name("kerberos-auth").build(), + spec: core::v1alpha1::AuthenticationClassSpec { + provider: core::v1alpha1::AuthenticationClassProvider::Kerberos( + kerberos::v1alpha1::AuthenticationProvider { + kerberos_secret_class: "kerberos-secret-class".to_string(), + }, + ), + }, + }]), + "tls".parse().unwrap(), + Some("tls".parse().unwrap()), + None, + ) + } + + #[test] + fn controller_role_mounts_pod_scoped_keytab_without_kcat_container() { + let mut pb = PodBuilder::new(); + let mut cb_kafka = ContainerBuilder::new("kafka").expect("valid container name"); + + add_kerberos_pod_config( + &kerberos_security(), + &KafkaRole::Controller, + None, + &mut cb_kafka, + &mut pb, + ) + .expect("kerberos pod config for controller role"); + + let pod = pb.build_template(); + let kerberos_volume = pod + .spec + .as_ref() + .and_then(|spec| spec.volumes.as_ref()) + .and_then(|volumes| volumes.iter().find(|v| v.name == "kerberos")) + .expect("kerberos volume must be present"); + let ephemeral = kerberos_volume + .ephemeral + .as_ref() + .expect("kerberos volume must be an ephemeral (secret-operator) volume"); + let annotations = ephemeral + .volume_claim_template + .as_ref() + .and_then(|t| t.metadata.annotations.as_ref()) + .expect("volume claim template must carry secrets.stackable.tech annotations"); + assert!( + !annotations.contains_key("secrets.stackable.tech/scope"), + "controller keytab must be pod-scoped only, not listener-volume-scoped: {annotations:?}" + ); + + let kafka_container = cb_kafka.build(); + let env_names: Vec<_> = kafka_container + .env + .unwrap_or_default() + .into_iter() + .map(|e| e.name) + .collect(); + assert!(env_names.contains(&"KRB5_CONFIG".to_string())); + assert!(env_names.contains(&"KAFKA_OPTS".to_string())); + } +} +``` + +(This test exercises the new `Option<&mut ContainerBuilder>` signature before it exists — it will fail to compile, which counts as "fails".) + +- [ ] **Step 2: Run test to verify it fails** + +Run: `cd rust && cargo test -p stackable-kafka-operator-binary controller::build::kerberos::` +Expected: FAIL to compile — `add_kerberos_pod_config` still takes `&mut ContainerBuilder`, not `Option<...>`, and `.with_listener_volume_scope` calls exist unconditionally so the "must be pod-scoped" assertion would fail once it *does* compile against the old body. + +- [ ] **Step 3: Implement** + +Replace the full body of `rust/operator-binary/src/controller/build/kerberos.rs` (keep the existing `Error` enum and imports, add `role::KafkaRole` usage which is already imported) with: + +```rust +pub fn add_kerberos_pod_config( + kafka_security: &ValidatedKafkaSecurity, + role: &KafkaRole, + cb_kcat_prober: Option<&mut ContainerBuilder>, + cb_kafka: &mut ContainerBuilder, + pb: &mut PodBuilder, +) -> Result<(), Error> { + if let Some(kerberos_secret_class) = kafka_security.kerberos_secret_class() { + let mut volume_builder = SecretOperatorVolumeSourceBuilder::new( + kerberos_secret_class, + // We need both public (krb5.conf) and private (keytab) parts. + SecretClassVolumeProvisionParts::PublicPrivate, + ); + volume_builder = match role { + // Brokers are exposed through listener-operator `Listener` volumes (the client + // and bootstrap listeners); the keytab principal must cover both. + KafkaRole::Broker => volume_builder + .with_listener_volume_scope(LISTENER_BROKER_VOLUME_NAME) + .with_listener_volume_scope(LISTENER_BOOTSTRAP_VOLUME_NAME), + // KRaft controllers have no listener-operator `Listener` volume (see + // `controller/build/mod.rs`, "Only broker role groups get a bootstrap Listener"): + // they're only reachable through their own StatefulSet pod DNS name, so the keytab + // must be pod-scoped, matching how the controller's internal TLS cert is provisioned + // in `add_controller_volume_and_volume_mounts`. + KafkaRole::Controller => volume_builder.with_pod_scope(), + }; + let kerberos_secret_operator_volume = volume_builder + .with_kerberos_service_name(role.kerberos_service_name()) + .build() + .context(KerberosSecretVolumeSnafu)?; + pb.add_volume( + VolumeBuilder::new("kerberos") + .ephemeral(kerberos_secret_operator_volume) + .build(), + ) + .context(AddVolumeSnafu)?; + + let mut containers: Vec<&mut ContainerBuilder> = vec![cb_kafka]; + if let Some(cb_kcat_prober) = cb_kcat_prober { + containers.push(cb_kcat_prober); + } + for cb in containers { + cb.add_volume_mount("kerberos", STACKABLE_KERBEROS_DIR) + .context(AddVolumeMountSnafu)?; + cb.add_env_var("KRB5_CONFIG", STACKABLE_KERBEROS_KRB5_PATH); + cb.add_env_var( + "KAFKA_OPTS", + format!("-Djava.security.auth.login.config=/tmp/jaas.properties -Djava.security.krb5.conf={STACKABLE_KERBEROS_KRB5_PATH}",), + ); + } + } + + Ok(()) +} +``` + +Update the single existing call site in `rust/operator-binary/src/controller/build/resource/statefulset.rs:243-250` (inside `build_broker_rolegroup_statefulset`) from: + +```rust + if kafka_security.has_kerberos_enabled() { + add_kerberos_pod_config( + kafka_security, + kafka_role, + &mut cb_kcat_prober, + &mut cb_kafka, + &mut pod_builder, + ) + .context(AddKerberosConfigSnafu)?; + } +``` + +to: + +```rust + if kafka_security.has_kerberos_enabled() { + add_kerberos_pod_config( + kafka_security, + kafka_role, + Some(&mut cb_kcat_prober), + &mut cb_kafka, + &mut pod_builder, + ) + .context(AddKerberosConfigSnafu)?; + } +``` + +Note: `volume_builder.with_pod_scope()`/`.with_listener_volume_scope(...)` on `SecretOperatorVolumeSourceBuilder` consume and return `Self` by value in this codebase's builder style (matches existing use in `controller/build/security.rs:319` and `kerberos.rs:55-56`) — if the actual builder signatures differ (e.g. `&mut self`), adjust the `match` arms to mutate in place instead of reassigning; check `stackable_operator::builder::pod::volume::SecretOperatorVolumeSourceBuilder`'s method signatures before finalizing. + +- [ ] **Step 4: Run test to verify it passes** + +Run: `cd rust && cargo test -p stackable-kafka-operator-binary controller::build::kerberos::` +Expected: PASS. + +Run also the broader build to catch the call-site change: `cd rust && cargo build -p stackable-kafka-operator-binary` +Expected: builds cleanly (only one call site to update, per Task research). + +- [ ] **Step 5: Commit** + +```bash +git add rust/operator-binary/src/controller/build/kerberos.rs rust/operator-binary/src/controller/build/resource/statefulset.rs +git commit -m "feat: support pod-scoped Kerberos keytabs for roles without a kcat prober container" +``` + +--- + +### Task 4: Wire Kerberos into the KRaft controller StatefulSet + +**Files:** +- Modify: `rust/operator-binary/src/controller/build/resource/statefulset.rs` — `build_controller_rolegroup_statefulset` (~line 455-641) +- Test: `rust/operator-binary/src/controller/build/resource/statefulset.rs` (add a new test near any existing `build_controller_rolegroup_statefulset` tests, or create one if none exist — check the file for the existing test module name first) + +**Interfaces:** +- Consumes: `add_kerberos_pod_config` (new signature from Task 3), `kafka_security: &ValidatedKafkaSecurity` (already a parameter of this function). +- Produces: when `kafka_security.has_kerberos_enabled()`, the controller pod template gets a `kerberos` volume, a `kerberos` volume mount on the `kafka` container, and `KRB5_CONFIG`/`KAFKA_OPTS` env vars on that container — consumed by Task 5's command changes (`KAFKA_OPTS` must point at the `/tmp/jaas.properties` file Task 5 populates). + +- [ ] **Step 1: Write the failing test** + +Find the existing controller StatefulSet test fixture in `rust/operator-binary/src/controller/build/resource/statefulset.rs` (search for `fn build_controller_rolegroup_statefulset` usage inside `#[cfg(test)] mod tests`) and add: + +```rust + #[test] + fn controller_statefulset_mounts_kerberos_when_enabled() { + let sts = build_controller_rolegroup_statefulset(/* ...use the same fixture helper the other controller statefulset tests use, with a kerberos()-enabled ValidatedKafkaSecurity... */) + .expect("controller statefulset build"); + + let kafka_container = sts + .spec + .expect("statefulset spec") + .template + .spec + .expect("pod spec") + .containers + .into_iter() + .find(|c| c.name == "kafka") + .expect("kafka container"); + + let env_names: Vec<_> = kafka_container + .env + .unwrap_or_default() + .into_iter() + .map(|e| e.name) + .collect(); + assert!(env_names.contains(&"KRB5_CONFIG".to_string())); + assert!(env_names.contains(&"KAFKA_OPTS".to_string())); + + let mount_names: Vec<_> = kafka_container + .volume_mounts + .unwrap_or_default() + .into_iter() + .map(|m| m.name) + .collect(); + assert!(mount_names.contains(&"kerberos".to_string())); + } +``` + +Adjust the call to `build_controller_rolegroup_statefulset(...)` to match whatever fixture-building helper the surrounding tests already use (e.g. `validated_cluster`/`minimal_kafka` helpers seen in `controller/build/properties/listener.rs`'s tests) — reuse the existing pattern in this file rather than inventing a new one. + +- [ ] **Step 2: Run test to verify it fails** + +Run: `cd rust && cargo test -p stackable-kafka-operator-binary controller::build::resource::statefulset::tests::controller_statefulset_mounts_kerberos_when_enabled` +Expected: FAIL — no `kerberos` env vars/volume mount present on the controller's `kafka` container. + +- [ ] **Step 3: Implement** + +In `build_controller_rolegroup_statefulset`, after the existing call to `add_controller_volume_and_volume_mounts` (`controller/build/security.rs`, already invoked in this function) and before the pod template is finalized, add: + +```rust + if kafka_security.has_kerberos_enabled() { + add_kerberos_pod_config( + kafka_security, + kafka_role, + None, + &mut cb_kafka, + &mut pod_builder, + ) + .context(AddKerberosConfigSnafu)?; + } +``` + +using whatever the existing local variable names are for `cb_kafka` / `pod_builder` in this function (match the broker function's naming, which this function generally mirrors). Make sure `add_kerberos_pod_config` and the `AddKerberosConfigSnafu` context (already defined as an `Error` variant used by the broker path) are in scope/imported in this file — they already are, since the broker branch uses them. + +- [ ] **Step 4: Run test to verify it passes** + +Run: `cd rust && cargo test -p stackable-kafka-operator-binary controller::build::resource::statefulset::` +Expected: PASS for all statefulset tests (including the new one and the pre-existing controller/broker ones, unaffected since they don't enable Kerberos). + +- [ ] **Step 5: Commit** + +```bash +git add rust/operator-binary/src/controller/build/resource/statefulset.rs +git commit -m "feat: mount Kerberos keytab into KRaft controller pods when enabled" +``` + +--- + +### Task 5: Generate a `controller.KafkaServer` JAAS section + +The static JAAS file (`jaas.properties`, generated by `jaas_config_file()`) needs a `controller.KafkaServer` login context for the `CONTROLLER` listener, matching the existing `bootstrap.KafkaServer` / `client.KafkaServer` naming convention. The principal differs by role: +- **Broker** pods act as SASL *clients* connecting out to controllers; their keytab (Task 3) only contains principals for the broker/bootstrap listener addresses, so the broker's `controller.KafkaServer` principal must reuse the same broker address already used for `client.KafkaServer`. +- **Controller** pods act as SASL *servers* (and peers to each other); their keytab (Task 3, `.with_pod_scope()`) is bound to their own pod FQDN, so their `controller.KafkaServer` principal must use that pod's own FQDN — the exact same env-var template already used for `KAFKA_LISTENERS` in `controller_properties.rs:48-51`. + +**Files:** +- Modify: `rust/operator-binary/src/controller/build/resource/config_map.rs` — `jaas_config_file()` (~line 199-229) and its one call site in `build_rolegroup_config_map` (~line 172) +- Test: same file (existing `#[cfg(test)] mod tests` block, ~line 231) + +**Interfaces:** +- Consumes: `KafkaRole` (new parameter), `crate::crd::role::KafkaRole`. +- Produces: `jaas_config_file(is_kerberos_enabled: bool, role: &KafkaRole) -> String` — signature changes by adding the `role` parameter; the one call site in `build_rolegroup_config_map` (which already computes `let role = validated_rg.config.config.kafka_role();`, per prior investigation) passes it through. + +- [ ] **Step 1: Write the failing tests** + +Replace the existing test module in `rust/operator-binary/src/controller/build/resource/config_map.rs` (currently just `jaas_config_file_empty_without_kerberos`) with: + +```rust +#[cfg(test)] +mod tests { + use super::jaas_config_file; + use crate::crd::role::KafkaRole; + + #[test] + fn jaas_config_file_empty_without_kerberos() { + assert_eq!(jaas_config_file(false, &KafkaRole::Broker), ""); + assert_eq!(jaas_config_file(false, &KafkaRole::Controller), ""); + } + + #[test] + fn jaas_config_file_broker_has_controller_section_using_broker_address() { + let jaas = jaas_config_file(true, &KafkaRole::Broker); + assert!(jaas.contains("controller.KafkaServer {")); + assert!(jaas.contains("kafka/${file:UTF-8:/stackable/listener-broker/default-address/address}@${env:KERBEROS_REALM}")); + } + + #[test] + fn jaas_config_file_controller_has_controller_section_using_pod_fqdn() { + let jaas = jaas_config_file(true, &KafkaRole::Controller); + assert!(jaas.contains("controller.KafkaServer {")); + assert!(jaas.contains( + "kafka/${env:POD_NAME}.${env:ROLEGROUP_HEADLESS_SERVICE_NAME}.${env:NAMESPACE}.svc.${env:CLUSTER_DOMAIN}@${env:KERBEROS_REALM}" + )); + // Controllers have no listener-operator Listener volume, so the broker-only sections + // must not appear in their JAAS file. + assert!(!jaas.contains("bootstrap.KafkaServer")); + assert!(!jaas.contains("client.KafkaServer")); + } +} +``` + +(Double-check the exact string produced by `node_address_cmd(STACKABLE_LISTENER_BROKER_DIR)` — defined in `crd/listener.rs:195-197` as `${{file:UTF-8:{directory}/default-address/address}}` — against `STACKABLE_LISTENER_BROKER_DIR`'s actual value before asserting the literal string; adjust the assertion to match exactly rather than guessing.) + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `cd rust && cargo test -p stackable-kafka-operator-binary controller::build::resource::config_map::tests::` +Expected: FAIL to compile (wrong arity) then, once fixed to compile, FAIL on missing `controller.KafkaServer` content. + +- [ ] **Step 3: Implement** + +Replace `jaas_config_file` in `rust/operator-binary/src/controller/build/resource/config_map.rs`: + +```rust +// Generate JAAS configuration file for Kerberos authentication +// or an empty string if Kerberos is not enabled. +// See https://docs.oracle.com/javase/8/docs/technotes/guides/security/jgss/tutorials/LoginConfigFile.html +fn jaas_config_file(is_kerberos_enabled: bool, role: &KafkaRole) -> String { + if !is_kerberos_enabled { + return String::new(); + } + + // Broker pods reach the CONTROLLER listener as SASL clients; the only principals in their + // keytab (see `add_kerberos_pod_config`) are for the broker/bootstrap listener addresses, so + // the CONTROLLER section must reuse the same address as `client.KafkaServer`. + // Controller pods have no listener-operator Listener volume; their keytab is pod-scoped, so + // the CONTROLLER section must use their own pod FQDN — the same template already used for + // `KAFKA_LISTENERS` in `controller_properties.rs`. + let controller_principal_address = match role { + KafkaRole::Broker => node_address_cmd(STACKABLE_LISTENER_BROKER_DIR), + KafkaRole::Controller => { + "${env:POD_NAME}.${env:ROLEGROUP_HEADLESS_SERVICE_NAME}.${env:NAMESPACE}.svc.${env:CLUSTER_DOMAIN}" + .to_string() + } + }; + + let controller_section = formatdoc! {" + controller.KafkaServer {{ + com.sun.security.auth.module.Krb5LoginModule required + useKeyTab=true + storeKey=true + isInitiator=false + keyTab=\"/stackable/kerberos/keytab\" + principal=\"kafka/{controller_principal_address}@${{env:KERBEROS_REALM}}\"; + }}; + ", + }; + + match role { + KafkaRole::Controller => controller_section, + KafkaRole::Broker => formatdoc! {" + bootstrap.KafkaServer {{ + com.sun.security.auth.module.Krb5LoginModule required + useKeyTab=true + storeKey=true + isInitiator=false + keyTab=\"/stackable/kerberos/keytab\" + principal=\"kafka/{bootstrap_address}@${{env:KERBEROS_REALM}}\"; + }}; + + client.KafkaServer {{ + com.sun.security.auth.module.Krb5LoginModule required + useKeyTab=true + storeKey=true + isInitiator=false + keyTab=\"/stackable/kerberos/keytab\" + principal=\"kafka/{broker_address}@${{env:KERBEROS_REALM}}\"; + }}; + + {controller_section} + ", + bootstrap_address = node_address_cmd(STACKABLE_LISTENER_BOOTSTRAP_DIR), + broker_address = node_address_cmd(STACKABLE_LISTENER_BROKER_DIR), + }, + } +} +``` + +Update the one call site in `build_rolegroup_config_map` (~line 172) from: + +```rust + cm_builder.add_data(ConfigFileName::Jaas.to_string(), jaas_config_file(is_kerberos_enabled)); +``` + +(or whatever the exact current call looks like — grep for `jaas_config_file(` to get the precise line) to pass `&role` (the `role` binding already computed at line 74 of this file, per prior investigation): + +```rust + cm_builder.add_data( + ConfigFileName::Jaas.to_string(), + jaas_config_file(is_kerberos_enabled, &role), + ); +``` + +Import `crate::crd::role::KafkaRole` at the top of the file if not already imported. + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `cd rust && cargo test -p stackable-kafka-operator-binary controller::build::resource::config_map::` +Expected: PASS. + +- [ ] **Step 5: Commit** + +```bash +git add rust/operator-binary/src/controller/build/resource/config_map.rs +git commit -m "feat: generate a controller.KafkaServer JAAS section for the CONTROLLER listener" +``` + +--- + +### Task 6: Wire `KERBEROS_REALM` export and JAAS templating into the controller startup command + +**Files:** +- Modify: `rust/operator-binary/src/controller/build/command.rs` — `controller_kafka_container_command` (~line 159-188) +- Modify: call site of `controller_kafka_container_command` in `rust/operator-binary/src/controller/build/resource/statefulset.rs` (`build_controller_rolegroup_statefulset`) +- Test: `rust/operator-binary/src/controller/build/command.rs` (existing `#[cfg(test)] mod tests`, ~line 231 area, or add one for this function if none exists yet — check first) + +**Interfaces:** +- Consumes: `ValidatedKafkaSecurity` (new parameter), `STACKABLE_KERBEROS_KRB5_PATH` (already imported in this file), `ConfigFileName::Jaas` (already imported). +- Produces: `controller_kafka_container_command(kafka_security: &ValidatedKafkaSecurity, controller_descriptors: Vec, product_version: &str) -> String` — signature changes by adding `kafka_security` as the first parameter, matching `broker_kafka_container_commands`'s existing parameter order/style. + +- [ ] **Step 1: Write the failing test** + +Add to `rust/operator-binary/src/controller/build/command.rs`'s test module: + +```rust + #[test] + fn controller_command_exports_kerberos_realm_and_templates_jaas_when_enabled() { + let command = controller_kafka_container_command(&kerberos_security(), vec![], "4.1.1"); + assert!(command.contains("export KERBEROS_REALM=")); + assert!(command.contains(&format!("cp {}/jaas.properties /tmp/jaas.properties", STACKABLE_CONFIG_DIR))); + assert!(command.contains("config-utils template /tmp/jaas.properties")); + } + + #[test] + fn controller_command_skips_kerberos_setup_when_disabled() { + let command = controller_kafka_container_command(&plaintext_security(), vec![], "4.1.1"); + assert!(!command.contains("KERBEROS_REALM")); + assert!(!command.contains("jaas.properties")); + } +``` + +Add the two small fixtures next to these tests, matching the pattern used elsewhere in this codebase (e.g. `controller/build/security.rs`'s `kerberos()`/`plaintext()` fixtures — reuse `ValidatedKafkaSecurity::new(...)` the same way): + +```rust + fn kerberos_security() -> ValidatedKafkaSecurity { /* same body as controller/build/security.rs's kerberos() fixture */ } + fn plaintext_security() -> ValidatedKafkaSecurity { /* same body as controller/build/security.rs's plaintext() fixture */ } +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `cd rust && cargo test -p stackable-kafka-operator-binary controller::build::command::tests::controller_command` +Expected: FAIL to compile (extra argument not accepted yet). + +- [ ] **Step 3: Implement** + +Replace `controller_kafka_container_command` in `rust/operator-binary/src/controller/build/command.rs`: + +```rust +pub fn controller_kafka_container_command( + kafka_security: &ValidatedKafkaSecurity, + controller_descriptors: Vec, + product_version: &str, +) -> String { + formatdoc! {" + {BASH_TRAP_FUNCTIONS} + {remove_vector_shutdown_file_command} + prepare_signal_handlers + containerdebug --output={STACKABLE_LOG_DIR}/containerdebug-state.json --loop & + {set_realm_env} + + POD_INDEX=$(echo \"$POD_NAME\" | grep -oE '[0-9]+$') + export REPLICA_ID=$((POD_INDEX+NODE_ID_OFFSET)) + + cp {config_dir}/{properties_file} /tmp/{properties_file} + + config-utils template /tmp/{properties_file} + + {jaas_setup} + + bin/kafka-storage.sh format --cluster-id \"$KAFKA_CLUSTER_ID\" --config /tmp/{properties_file} --ignore-formatted {initial_controller_command} + bin/kafka-server-start.sh /tmp/{properties_file} & + + wait_for_termination $! + {create_vector_shutdown_file_command} + ", + remove_vector_shutdown_file_command = remove_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(), + }, + config_dir = STACKABLE_CONFIG_DIR, + properties_file = ConfigFileName::ControllerProperties, + jaas_setup = match kafka_security.has_kerberos_enabled() { + true => formatdoc! {" + cp {config_dir}/{jaas_file} /tmp/{jaas_file} + config-utils template /tmp/{jaas_file}", + config_dir = STACKABLE_CONFIG_DIR, + jaas_file = ConfigFileName::Jaas, + }, + false => "".to_string(), + }, + initial_controller_command = initial_controllers_command(&controller_descriptors, product_version), + create_vector_shutdown_file_command = create_vector_shutdown_file_command(STACKABLE_LOG_DIR) + } +} +``` + +Add `use crate::controller::security::ValidatedKafkaSecurity;` to this file's imports if not already present (it's already imported per the top of the file, alongside `copy_opa_tls_cert_command`). + +Update the call site in `rust/operator-binary/src/controller/build/resource/statefulset.rs` (`build_controller_rolegroup_statefulset`), from: + +```rust +controller_kafka_container_command(controller_descriptors, product_version) +``` + +to: + +```rust +controller_kafka_container_command(kafka_security, controller_descriptors, product_version) +``` + +(exact argument names per the surrounding code; `kafka_security` is already a parameter of `build_controller_rolegroup_statefulset`). + +- [ ] **Step 4: Run test to verify it passes** + +Run: `cd rust && cargo test -p stackable-kafka-operator-binary controller::build::command::` +Expected: PASS. + +Run: `cd rust && cargo build -p stackable-kafka-operator-binary` +Expected: builds cleanly. + +- [ ] **Step 5: Commit** + +```bash +git add rust/operator-binary/src/controller/build/command.rs rust/operator-binary/src/controller/build/resource/statefulset.rs +git commit -m "feat: export KERBEROS_REALM and template jaas.properties in the KRaft controller startup command" +``` + +--- + +### Task 7: Full-stack Rust verification + +**Files:** none (verification only) + +- [ ] **Step 1: Run the full unit test suite** + +Run: `cd rust && cargo test -p stackable-kafka-operator-binary` +Expected: PASS, including every test touched in Tasks 1–6 plus all pre-existing ones (nothing about plaintext/TLS/ZooKeeper-mode/non-Kerberos KRaft behavior should have changed). + +- [ ] **Step 2: Run clippy** + +Run: `cd rust && cargo clippy -p stackable-kafka-operator-binary --all-targets -- -D warnings` +Expected: no warnings. + +- [ ] **Step 3: Regenerate CRDs/docs if the build pipeline requires it** + +Run whatever this repo's Makefile/justfile target regenerates generated artifacts (check `Makefile`/`justfile` for a `regenerate-charts`/`crd` target) — this task touches no CRD fields, so this step should be a no-op, but confirm no diff appears in `deploy/helm/kafka-operator/crds/crds.yaml`. + +- [ ] **Step 4: Commit if regeneration produced any diff** + +```bash +git add -A +git commit -m "chore: regenerate generated artifacts" +``` +(Skip this commit entirely if step 3 produced no diff.) + +--- + +### Task 8: kuttl integration test — Kerberos-secured KRaft cluster + +**Files:** +- Inspect: `tests/templates/kuttl/kerberos/` (existing Kerberos smoke test, ZooKeeper-mode) and `tests/templates/kuttl/smoke-kraft/` (existing KRaft smoke test, no Kerberos) to reuse their KDC-deployment and cluster-manifest boilerplate. +- Create: `tests/templates/kuttl/kraft-kerberos/` (new test case directory; mirror the structure of `smoke-kraft` and `kerberos`, e.g. `00-assert.yaml`/`00-install-krb5-kdc.yaml`, a `KafkaCluster` manifest with both `spec.controllers` and a Kerberos `AuthenticationClass`, and produce/consume assertions). +- Modify: `tests/test-definition.yaml` — register the new test case in the `kafka-kraft` (or equivalent) dimension list, following the existing `kafka-kraft`/`operations-kraft`/`smoke-kraft` entries. + +**Interfaces:** none (integration test, no Rust interfaces). + +- [ ] **Step 1: Copy the existing `smoke-kraft` test case as a starting point** + +```bash +cp -r tests/templates/kuttl/smoke-kraft tests/templates/kuttl/kraft-kerberos +``` + +- [ ] **Step 2: Add Kerberos KDC deployment and AuthenticationClass** + +Copy the MIT KDC deployment manifest and the `AuthenticationClass`/`SecretClass` (`kerberos-kafka` or similar) used in `tests/templates/kuttl/kerberos/` into the new `kraft-kerberos` test case's early numbered step (e.g. `00-install-krb5-kdc.yaml`), and reference that `AuthenticationClass` from the `KafkaCluster` manifest's `spec.clusterConfig.authentication` (following whatever field path the existing `kerberos` test case uses). + +- [ ] **Step 3: Keep the KRaft `spec.controllers` block from `smoke-kraft`** + +The `KafkaCluster` manifest should end up with both `spec.controllers.roleGroups` (from `smoke-kraft`) and the Kerberos `authentication` entry (from `kerberos`) present simultaneously — this is the actual scenario under test. + +- [ ] **Step 4: Reuse the existing produce/consume assertion steps** + +Copy the numbered steps that create a topic and produce/consume test messages using `kafka-topics.sh`/`kafka-producer-perf-test.sh`/`kafka-console-consumer.sh` with `--command-config`/`--producer.config`/`--consumer.config` pointing at the Kerberos `client.properties` from `tests/templates/kuttl/kerberos/`, adjusted to target the KRaft cluster's bootstrap service name. + +- [ ] **Step 5: Register the test case** + +In `tests/test-definition.yaml`, add `kraft-kerberos` alongside the existing `smoke-kraft`/`operations-kraft` dimension entries (lines ~78-84/104 per prior investigation), so it's picked up by `stackablectl` / CI test dimension generation the same way. + +- [ ] **Step 6: Run the test locally** + +Run: `./scripts/run_tests.sh --test-suite kraft-kerberos` (or whatever this repo's actual test-runner invocation is — check `tests/README.md` for the exact command) against a local kind/k3d cluster with the Kerberos operator and secret-operator installed. +Expected: PASS — controllers form quorum over Kerberos-authenticated `CONTROLLER` traffic, brokers join, topic create/produce/consume succeed end-to-end. + +- [ ] **Step 7: Commit** + +```bash +git add tests/templates/kuttl/kraft-kerberos tests/test-definition.yaml +git commit -m "test: add kuttl integration test for Kerberos-secured KRaft clusters" +``` + +--- + +### Task 9: Documentation updates + +**Files:** +- Modify: `docs/modules/kafka/pages/usage-guide/kraft-controller.adoc:90-94` (Known Issues) +- Modify: `docs/modules/kafka/partials/supported-versions.adoc:5-15` (experimental caveats) +- Modify: `CHANGELOG.md` (new entry) + +**Interfaces:** none (docs only). + +- [ ] **Step 1: Update the KRaft "Known Issues" section** + +In `docs/modules/kafka/pages/usage-guide/kraft-controller.adoc`, replace: + +```asciidoc +== Known Issues + +* Automatic migration from Apache ZooKeeper to KRaft is not supported. +* Scaling controller replicas might lead to unstable clusters. +* Kerberos is currently not supported for KRaft in all versions. +``` + +with: + +```asciidoc +== Known Issues + +* Automatic migration from Apache ZooKeeper to KRaft is not supported. +* Scaling controller replicas might lead to unstable clusters. +``` + +and add a short new subsection documenting Kerberos support, e.g. right after `=== Overrides`: + +```asciidoc +=== Kerberos + +Kerberos authentication is supported for KRaft clusters: enabling a Kerberos `AuthenticationClass` +secures the `CLIENT`, `INTERNAL` and `CONTROLLER` listeners alike, including controller-to-controller +and broker-to-controller Raft RPC traffic. + +NOTE: SASL/SCRAM is not supported for the controller listener by Apache Kafka itself +(https://issues.apache.org/jira/browse/KAFKA-15513[KAFKA-15513]); this does not affect Kerberos +(GSSAPI), which authenticates against the external KDC rather than Kafka-internal credential storage. +``` + +- [ ] **Step 2: Update the supported-versions matrix** + +In `docs/modules/kafka/partials/supported-versions.adoc`, remove the "Kerberos authentication is not tested yet." bullet for the affected versions (keep "Controller scaling is not reliable." and "Service exposition is not definitive." as-is unless this plan's implementation also happens to address them, which it does not). + +- [ ] **Step 3: Add a CHANGELOG entry** + +In `CHANGELOG.md`, under the `### Added` (or equivalent) section for the in-progress release, add: + +```markdown +- Kerberos authentication now works with KRaft controllers (`spec.controllers`), securing the + `CONTROLLER` listener used for broker/controller and controller/controller Raft RPC traffic ([#]). +``` + +- [ ] **Step 4: Commit** + +```bash +git add docs/modules/kafka/pages/usage-guide/kraft-controller.adoc docs/modules/kafka/partials/supported-versions.adoc CHANGELOG.md +git commit -m "docs: document Kerberos support for KRaft controllers" +``` + +--- + +## Self-Review Notes + +- **Spec coverage:** the original question was "how does Kerberos authentication work with coordinators [KRaft controllers]" — answered in Research Findings (Kafka supports `SASL_SSL` on `CONTROLLER`; the real limitation is SCRAM, not GSSAPI). Every concrete code gap found while answering that question (listener protocol map, `sasl.mechanism.controller.protocol`, keytab mounting, JAAS section, startup command, docs) has a corresponding task (Tasks 1–6, 9). Tasks 7–8 cover verification and end-to-end proof. +- **Open risk to flag to a reviewer before Task 3 lands:** the exact mutability/ownership signature of `SecretOperatorVolumeSourceBuilder::with_pod_scope()` / `with_listener_volume_scope()` in the pinned `stackable-operator` crate version should be double-checked (Step 3 of Task 3 already calls this out) — if either takes `&mut self` instead of consuming `self`, the `match` arms in that task need `let mut volume_builder = ...; match role { ... volume_builder.with_pod_scope(); ... }` instead of reassignment. +- **Open risk to flag to a reviewer before Task 8:** this plan assumes the secret-operator, when given `.with_pod_scope()` + `.with_kerberos_service_name(...)` with no listener-volume scope, mints a keytab principal bound to the pod's own StatefulSet-derived FQDN (the same assumption the existing pod-scoped internal-TLS cert relies on for its SANs). This should hold given the existing pattern for TLS certs, but is worth an explicit smoke-test check (Task 8, Step 6) before considering the feature done — if it doesn't hold, Task 3's assumption about which hostname ends up in the keytab needs revisiting together with Task 5's principal templating. From 5939302bd2dc368b919e7ecad77b54607243ae80 Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Fri, 7 Aug 2026 15:17:05 +0200 Subject: [PATCH 02/14] feat: allow SASL_SSL on the KRaft CONTROLLER listener when Kerberos is enabled Co-Authored-By: Claude Sonnet 5 --- .../src/controller/build/properties/listener.rs | 12 +++++++++--- rust/operator-binary/src/crd/listener.rs | 7 ++----- 2 files changed, 11 insertions(+), 8 deletions(-) diff --git a/rust/operator-binary/src/controller/build/properties/listener.rs b/rust/operator-binary/src/controller/build/properties/listener.rs index 3ba733bf..acc63255 100644 --- a/rust/operator-binary/src/controller/build/properties/listener.rs +++ b/rust/operator-binary/src/controller/build/properties/listener.rs @@ -108,8 +108,14 @@ pub fn get_kafka_listener_config( port: kafka_security.internal_port().to_string(), }); listener_security_protocol_map.insert(KafkaListenerName::Internal, KafkaListenerProtocol::Ssl); - listener_security_protocol_map - .insert(KafkaListenerName::Controller, KafkaListenerProtocol::Ssl); + listener_security_protocol_map.insert( + KafkaListenerName::Controller, + if kafka_security.has_kerberos_enabled() { + KafkaListenerProtocol::SaslSsl + } else { + KafkaListenerProtocol::Ssl + }, + ); // BOOTSTRAP if kafka_security.has_kerberos_enabled() { @@ -492,7 +498,7 @@ mod tests { bootstrap_name = KafkaListenerName::Bootstrap, bootstrap_protocol = KafkaListenerProtocol::SaslSsl, controller_name = KafkaListenerName::Controller, - controller_protocol = KafkaListenerProtocol::Ssl, + controller_protocol = KafkaListenerProtocol::SaslSsl, ) ); } diff --git a/rust/operator-binary/src/crd/listener.rs b/rust/operator-binary/src/crd/listener.rs index 7aabadad..8e014d14 100644 --- a/rust/operator-binary/src/crd/listener.rs +++ b/rust/operator-binary/src/crd/listener.rs @@ -58,13 +58,10 @@ pub enum KafkaListenerName { /// This listener is defined when Kraft mode is enabled. /// It is responsible for broker/controller as well as controller/controller communications /// and therefore it is present on *both* brokers and controller properties files. - /// The only protocol used is SSL. + /// The protocol used is SSL, or SASL_SSL when Kerberos is enabled. /// The advertised host names are FQDN pod names of the controllers. /// - /// Notes: - /// - /// - there is no listener for client/controller communication - /// - this listener does not support SSL_SASL. + /// Note: there is no listener for client/controller communication. #[strum(serialize = "CONTROLLER")] Controller, } From 798f810273b80290740effd22d0cb34db0917526 Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Fri, 7 Aug 2026 15:21:26 +0200 Subject: [PATCH 03/14] feat: set sasl.mechanism.controller.protocol for Kerberos-enabled clusters Co-Authored-By: Claude Sonnet 5 --- .../src/controller/build/security.rs | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/rust/operator-binary/src/controller/build/security.rs b/rust/operator-binary/src/controller/build/security.rs index e36191c1..77104d0c 100644 --- a/rust/operator-binary/src/controller/build/security.rs +++ b/rust/operator-binary/src/controller/build/security.rs @@ -49,6 +49,7 @@ const PROPERTY_SECURITY_PROTOCOL: &str = "security.protocol"; const PROPERTY_SASL_ENABLED_MECHANISMS: &str = "sasl.enabled.mechanisms"; const PROPERTY_SASL_KERBEROS_SERVICE_NAME: &str = "sasl.kerberos.service.name"; const PROPERTY_SASL_INTER_BROKER_MECHANISM: &str = "sasl.mechanism.inter.broker.protocol"; +const PROPERTY_SASL_CONTROLLER_MECHANISM: &str = "sasl.mechanism.controller.protocol"; const STACKABLE_TLS_KAFKA_INTERNAL_DIR: &str = "/stackable/tls-kafka-internal"; const STACKABLE_TLS_KAFKA_INTERNAL_VOLUME_NAME: &str = "tls-kafka-internal"; const STACKABLE_TLS_KAFKA_SERVER_DIR: &str = "/stackable/tls-kafka-server"; @@ -448,6 +449,10 @@ pub fn broker_config_settings(security: &ValidatedKafkaSecurity) -> BTreeMap BTreeMap PROPERTY_SASL_INTER_BROKER_MECHANISM.to_string(), SASL_MECHANISM_GSSAPI.to_string(), ); + config.insert( + PROPERTY_SASL_CONTROLLER_MECHANISM.to_string(), + SASL_MECHANISM_GSSAPI.to_string(), + ); tracing::debug!("Kerberos configs added: [{:#?}]", config); } @@ -933,6 +942,10 @@ mod tests { config.get("sasl.mechanism.inter.broker.protocol"), Some(&"GSSAPI".to_string()) ); + assert_eq!( + config.get("sasl.mechanism.controller.protocol"), + Some(&"GSSAPI".to_string()) + ); assert!(config.contains_key("listener.name.bootstrap.ssl.keystore.location")); } @@ -988,5 +1001,9 @@ mod tests { config.get("sasl.kerberos.service.name"), Some(&"kafka".to_string()) ); + assert_eq!( + config.get("sasl.mechanism.controller.protocol"), + Some(&"GSSAPI".to_string()) + ); } } From a9be223892f069d152b84aea2e48ef2e2eb953e6 Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Fri, 7 Aug 2026 15:28:33 +0200 Subject: [PATCH 04/14] feat: support pod-scoped Kerberos keytabs for roles without a kcat prober container add_kerberos_pod_config now takes cb_kcat_prober: Option<&mut ContainerBuilder> instead of a required reference, and scopes the Kerberos keytab volume based on the KafkaRole: Broker keeps the existing listener-volume scoping (client + bootstrap listeners), Controller gets a pod-scoped keytab since KRaft controllers have no listener-operator Listener volume. Updates the one existing call site in build_broker_rolegroup_statefulset to pass Some(&mut cb_kcat_prober). Adds a unit test covering the controller-role path (pod-scoped keytab, no kcat container, env vars still applied to the kafka container). Co-Authored-By: Claude Sonnet 5 --- .../src/controller/build/kerberos.rs | 121 ++++++++++++++++-- .../controller/build/resource/statefulset.rs | 2 +- 2 files changed, 112 insertions(+), 11 deletions(-) diff --git a/rust/operator-binary/src/controller/build/kerberos.rs b/rust/operator-binary/src/controller/build/kerberos.rs index 676c45b0..46c5842b 100644 --- a/rust/operator-binary/src/controller/build/kerberos.rs +++ b/rust/operator-binary/src/controller/build/kerberos.rs @@ -41,22 +41,37 @@ pub enum Error { pub fn add_kerberos_pod_config( kafka_security: &ValidatedKafkaSecurity, role: &KafkaRole, - cb_kcat_prober: &mut ContainerBuilder, + cb_kcat_prober: Option<&mut ContainerBuilder>, cb_kafka: &mut ContainerBuilder, pb: &mut PodBuilder, ) -> Result<(), Error> { if let Some(kerberos_secret_class) = kafka_security.kerberos_secret_class() { - // Mount keytab - let kerberos_secret_operator_volume = SecretOperatorVolumeSourceBuilder::new( + let mut volume_builder = SecretOperatorVolumeSourceBuilder::new( kerberos_secret_class, // 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_kerberos_service_name(role.kerberos_service_name()) - .build() - .context(KerberosSecretVolumeSnafu)?; + ); + match role { + // Brokers are exposed through listener-operator `Listener` volumes (the client + // and bootstrap listeners); the keytab principal must cover both. + KafkaRole::Broker => { + volume_builder + .with_listener_volume_scope(LISTENER_BROKER_VOLUME_NAME) + .with_listener_volume_scope(LISTENER_BOOTSTRAP_VOLUME_NAME); + } + // KRaft controllers have no listener-operator `Listener` volume (see + // `controller/build/mod.rs`, "Only broker role groups get a bootstrap Listener"): + // they're only reachable through their own StatefulSet pod DNS name, so the keytab + // must be pod-scoped, matching how the controller's internal TLS cert is provisioned + // in `add_controller_volume_and_volume_mounts`. + KafkaRole::Controller => { + volume_builder.with_pod_scope(); + } + }; + let kerberos_secret_operator_volume = volume_builder + .with_kerberos_service_name(role.kerberos_service_name()) + .build() + .context(KerberosSecretVolumeSnafu)?; pb.add_volume( VolumeBuilder::new("kerberos") .ephemeral(kerberos_secret_operator_volume) @@ -64,7 +79,11 @@ pub fn add_kerberos_pod_config( ) .context(AddVolumeSnafu)?; - for cb in [cb_kafka, cb_kcat_prober] { + let mut containers: Vec<&mut ContainerBuilder> = vec![cb_kafka]; + if let Some(cb_kcat_prober) = cb_kcat_prober { + containers.push(cb_kcat_prober); + } + for cb in containers { cb.add_volume_mount("kerberos", STACKABLE_KERBEROS_DIR) .context(AddVolumeMountSnafu)?; cb.add_env_var("KRB5_CONFIG", STACKABLE_KERBEROS_KRB5_PATH); @@ -77,3 +96,85 @@ pub fn add_kerberos_pod_config( Ok(()) } + +#[cfg(test)] +mod tests { + use stackable_operator::{ + builder::{meta::ObjectMetaBuilder, pod::container::ContainerBuilder}, + crd::authentication::{core, kerberos}, + }; + + use super::*; + use crate::crd::authentication::ResolvedAuthenticationClasses; + + fn kerberos_security() -> ValidatedKafkaSecurity { + ValidatedKafkaSecurity::new( + ResolvedAuthenticationClasses::new(vec![core::v1alpha1::AuthenticationClass { + metadata: ObjectMetaBuilder::new().name("kerberos-auth").build(), + spec: core::v1alpha1::AuthenticationClassSpec { + provider: core::v1alpha1::AuthenticationClassProvider::Kerberos( + kerberos::v1alpha1::AuthenticationProvider { + kerberos_secret_class: "kerberos-secret-class".to_string(), + }, + ), + }, + }]), + "tls".parse().unwrap(), + Some("tls".parse().unwrap()), + None, + ) + } + + #[test] + fn controller_role_mounts_pod_scoped_keytab_without_kcat_container() { + let mut pb = PodBuilder::new(); + let mut cb_kafka = ContainerBuilder::new("kafka").expect("valid container name"); + + add_kerberos_pod_config( + &kerberos_security(), + &KafkaRole::Controller, + None, + &mut cb_kafka, + &mut pb, + ) + .expect("kerberos pod config for controller role"); + + let pod = pb.build_template(); + let kerberos_volume = pod + .spec + .as_ref() + .and_then(|spec| spec.volumes.as_ref()) + .and_then(|volumes| volumes.iter().find(|v| v.name == "kerberos")) + .expect("kerberos volume must be present"); + let ephemeral = kerberos_volume + .ephemeral + .as_ref() + .expect("kerberos volume must be an ephemeral (secret-operator) volume"); + let annotations = ephemeral + .volume_claim_template + .as_ref() + .and_then(|t| t.metadata.as_ref()) + .and_then(|m| m.annotations.as_ref()) + .expect("volume claim template must carry secrets.stackable.tech annotations"); + // Pod-scoping (`with_pod_scope()`) is expressed as a `secrets.stackable.tech/scope: pod` + // annotation (same as the controller's internal TLS cert, see + // `add_controller_volume_and_volume_mounts`) -- it must not mention a listener volume. + assert_eq!( + annotations + .get("secrets.stackable.tech/scope") + .map(String::as_str), + Some("pod"), + "controller keytab must be pod-scoped only, not listener-volume-scoped: {annotations:?}" + ); + + let kafka_container = cb_kafka.build(); + let env_names: Vec<_> = kafka_container + .env + .unwrap_or_default() + .into_iter() + .map(|e| e.name) + .collect(); + assert!(env_names.contains(&"KRB5_CONFIG".to_string())); + assert!(env_names.contains(&"KAFKA_OPTS".to_string())); + } +} diff --git a/rust/operator-binary/src/controller/build/resource/statefulset.rs b/rust/operator-binary/src/controller/build/resource/statefulset.rs index 57104c50..704d5395 100644 --- a/rust/operator-binary/src/controller/build/resource/statefulset.rs +++ b/rust/operator-binary/src/controller/build/resource/statefulset.rs @@ -243,7 +243,7 @@ pub fn build_broker_rolegroup_statefulset( add_kerberos_pod_config( kafka_security, kafka_role, - &mut cb_kcat_prober, + Some(&mut cb_kcat_prober), &mut cb_kafka, &mut pod_builder, ) From 36a6992ad64e312393fb54a7adcf0d49f5072ab8 Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Fri, 7 Aug 2026 15:38:45 +0200 Subject: [PATCH 05/14] feat: mount Kerberos keytab into KRaft controller pods when enabled Wire the already-refactored add_kerberos_pod_config (Task 3) into build_controller_rolegroup_statefulset so KRaft controller pods get a pod-scoped kerberos volume, KRB5_CONFIG and KAFKA_OPTS on the kafka container when kafka_security.has_kerberos_enabled() is true. Broker pods already got this via build_broker_rolegroup_statefulset. Adds two unit tests covering the enabled and disabled cases. Co-Authored-By: Claude Sonnet 5 --- .../controller/build/resource/statefulset.rs | 200 ++++++++++++++++++ 1 file changed, 200 insertions(+) diff --git a/rust/operator-binary/src/controller/build/resource/statefulset.rs b/rust/operator-binary/src/controller/build/resource/statefulset.rs index 704d5395..d0dbf2bd 100644 --- a/rust/operator-binary/src/controller/build/resource/statefulset.rs +++ b/rust/operator-binary/src/controller/build/resource/statefulset.rs @@ -572,6 +572,17 @@ pub fn build_controller_rolegroup_statefulset( ) .context(AddVolumesAndVolumeMountsSnafu)?; + if kafka_security.has_kerberos_enabled() { + add_kerberos_pod_config( + kafka_security, + kafka_role, + None, + &mut cb_kafka, + &mut pod_builder, + ) + .context(AddKerberosConfigSnafu)?; + } + let kafka_container = cb_kafka.build(); pod_builder @@ -797,3 +808,192 @@ fn add_vector_container( )); } } + +#[cfg(test)] +mod tests { + use stackable_operator::{ + builder::meta::ObjectMetaBuilder, + crd::authentication::{core, kerberos}, + }; + + use super::*; + use crate::{ + controller::test_support::{minimal_kafka, validated_cluster}, + crd::authentication::ResolvedAuthenticationClasses, + }; + + /// A Kerberos-enabled [`ValidatedKafkaSecurity`], mirroring the fixture used by + /// `add_kerberos_pod_config`'s own tests (`controller/build/kerberos.rs`). + fn kerberos_security() -> ValidatedKafkaSecurity { + ValidatedKafkaSecurity::new( + ResolvedAuthenticationClasses::new(vec![core::v1alpha1::AuthenticationClass { + metadata: ObjectMetaBuilder::new().name("kerberos-auth").build(), + spec: core::v1alpha1::AuthenticationClassSpec { + provider: core::v1alpha1::AuthenticationClassProvider::Kerberos( + kerberos::v1alpha1::AuthenticationProvider { + kerberos_secret_class: "kerberos-secret-class".to_string(), + }, + ), + }, + }]), + "tls".parse().unwrap(), + Some("tls".parse().unwrap()), + None, + ) + } + + /// A KRaft cluster with one `controller` and one `broker` role group, resolved through the + /// real validate step (mirroring the fixtures in `controller/build/mod.rs`'s tests). The + /// `kafka_security` is swapped for a Kerberos-enabled one afterwards, since `validate()` + /// only resolves auth classes that are actually referenced from the cluster spec, and both + /// `ValidatedCluster::cluster_config` and `ValidatedKafkaSecurity` fields are public. + fn kraft_cluster_with_kerberos() -> ValidatedCluster { + let kafka = minimal_kafka( + r#" + apiVersion: kafka.stackable.tech/v1alpha1 + kind: KafkaCluster + metadata: + name: simple-kafka + namespace: default + uid: 12345678-1234-1234-1234-123456789012 + spec: + image: + productVersion: 3.9.2 + clusterConfig: + metadataManager: kraft + controllers: + roleGroups: + default: + replicas: 3 + brokers: + roleGroups: + default: + replicas: 3 + "#, + ); + let mut cluster = validated_cluster(&kafka); + cluster.cluster_config.kafka_security = kerberos_security(); + cluster + } + + #[test] + fn controller_statefulset_mounts_kerberos_when_enabled() { + let cluster = kraft_cluster_with_kerberos(); + let role_group_name: RoleGroupName = "default".parse().unwrap(); + let validated_rg = cluster + .role_group_configs + .get(&KafkaRole::Controller) + .expect("controller role group configs") + .get(&role_group_name) + .expect("default controller role group"); + + let sts = build_controller_rolegroup_statefulset( + &KafkaRole::Controller, + &role_group_name, + &cluster, + validated_rg, + ) + .expect("controller statefulset build"); + + let kafka_container = sts + .spec + .expect("statefulset spec") + .template + .spec + .expect("pod spec") + .containers + .into_iter() + .find(|c| c.name == "kafka") + .expect("kafka container"); + + let env_names: Vec<_> = kafka_container + .env + .unwrap_or_default() + .into_iter() + .map(|e| e.name) + .collect(); + assert!(env_names.contains(&"KRB5_CONFIG".to_string())); + assert!(env_names.contains(&"KAFKA_OPTS".to_string())); + + let mount_names: Vec<_> = kafka_container + .volume_mounts + .unwrap_or_default() + .into_iter() + .map(|m| m.name) + .collect(); + assert!(mount_names.contains(&"kerberos".to_string())); + } + + /// Non-Kerberos controller StatefulSets must be unaffected: no `kerberos` volume mount and + /// no Kerberos env vars on the `kafka` container. + #[test] + fn controller_statefulset_has_no_kerberos_when_disabled() { + let kafka = minimal_kafka( + r#" + apiVersion: kafka.stackable.tech/v1alpha1 + kind: KafkaCluster + metadata: + name: simple-kafka + namespace: default + uid: 12345678-1234-1234-1234-123456789012 + spec: + image: + productVersion: 3.9.2 + clusterConfig: + metadataManager: kraft + controllers: + roleGroups: + default: + replicas: 3 + brokers: + roleGroups: + default: + replicas: 3 + "#, + ); + let cluster = validated_cluster(&kafka); + let role_group_name: RoleGroupName = "default".parse().unwrap(); + let validated_rg = cluster + .role_group_configs + .get(&KafkaRole::Controller) + .expect("controller role group configs") + .get(&role_group_name) + .expect("default controller role group"); + + let sts = build_controller_rolegroup_statefulset( + &KafkaRole::Controller, + &role_group_name, + &cluster, + validated_rg, + ) + .expect("controller statefulset build"); + + let kafka_container = sts + .spec + .expect("statefulset spec") + .template + .spec + .expect("pod spec") + .containers + .into_iter() + .find(|c| c.name == "kafka") + .expect("kafka container"); + + let env_names: Vec<_> = kafka_container + .env + .unwrap_or_default() + .into_iter() + .map(|e| e.name) + .collect(); + assert!(!env_names.contains(&"KRB5_CONFIG".to_string())); + assert!(!env_names.contains(&"KAFKA_OPTS".to_string())); + + let mount_names: Vec<_> = kafka_container + .volume_mounts + .unwrap_or_default() + .into_iter() + .map(|m| m.name) + .collect(); + assert!(!mount_names.contains(&"kerberos".to_string())); + } +} From d9942ad8e4f443bcbd8d8d25189f7f10fad1bc9b Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Fri, 7 Aug 2026 15:43:37 +0200 Subject: [PATCH 06/14] feat: generate a controller.KafkaServer JAAS section for the CONTROLLER listener Broker pods reuse the broker listener address already used for client.KafkaServer, since their keytab only covers broker/bootstrap listener addresses. Controller pods use their own pod FQDN template (same as KAFKA_LISTENERS in controller_properties.rs), since they have no listener-operator Listener volume and their keytab is pod-scoped. Co-Authored-By: Claude Sonnet 5 --- .../controller/build/resource/config_map.rs | 93 +++++++++++++++---- 1 file changed, 74 insertions(+), 19 deletions(-) diff --git a/rust/operator-binary/src/controller/build/resource/config_map.rs b/rust/operator-binary/src/controller/build/resource/config_map.rs index 284a281b..c35e5cf6 100644 --- a/rust/operator-binary/src/controller/build/resource/config_map.rs +++ b/rust/operator-binary/src/controller/build/resource/config_map.rs @@ -23,7 +23,7 @@ use crate::{ crd::{ STACKABLE_LISTENER_BOOTSTRAP_DIR, STACKABLE_LISTENER_BROKER_DIR, listener::{KafkaListenerConfig, node_address_cmd}, - role::AnyConfig, + role::{AnyConfig, KafkaRole}, }, }; @@ -169,7 +169,7 @@ pub fn build_rolegroup_config_map( // and this tool currently doesn't support the JAAS login configuration format. .add_data( ConfigFileName::Jaas.to_string(), - jaas_config_file(kafka_security.has_kerberos_enabled()), + jaas_config_file(kafka_security.has_kerberos_enabled(), &role), ); tracing::debug!(?kafka_config, "Applied kafka config"); @@ -199,29 +199,60 @@ pub fn build_rolegroup_config_map( // Generate JAAS configuration file for Kerberos authentication // or an empty string if Kerberos is not enabled. // See https://docs.oracle.com/javase/8/docs/technotes/guides/security/jgss/tutorials/LoginConfigFile.html -fn jaas_config_file(is_kerberos_enabled: bool) -> String { - match is_kerberos_enabled { - false => String::new(), - true => formatdoc! {" - bootstrap.KafkaServer {{ - com.sun.security.auth.module.Krb5LoginModule required - useKeyTab=true - storeKey=true - isInitiator=false - keyTab=\"/stackable/kerberos/keytab\" - principal=\"kafka/{bootstrap_address}@${{env:KERBEROS_REALM}}\"; - }}; +fn jaas_config_file(is_kerberos_enabled: bool, role: &KafkaRole) -> String { + if !is_kerberos_enabled { + return String::new(); + } - client.KafkaServer {{ + // Broker pods reach the CONTROLLER listener as SASL clients; the only principals in their + // keytab (see `add_kerberos_pod_config`) are for the broker/bootstrap listener addresses, so + // the CONTROLLER section must reuse the same address as `client.KafkaServer`. + // Controller pods have no listener-operator Listener volume; their keytab is pod-scoped, so + // the CONTROLLER section must use their own pod FQDN — the same template already used for + // `KAFKA_LISTENERS` in `controller_properties.rs`. + let controller_principal_address = match role { + KafkaRole::Broker => node_address_cmd(STACKABLE_LISTENER_BROKER_DIR), + KafkaRole::Controller => { + "${env:POD_NAME}.${env:ROLEGROUP_HEADLESS_SERVICE_NAME}.${env:NAMESPACE}.svc.${env:CLUSTER_DOMAIN}" + .to_string() + } + }; + + let controller_section = formatdoc! {" + controller.KafkaServer {{ com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true storeKey=true isInitiator=false keyTab=\"/stackable/kerberos/keytab\" - principal=\"kafka/{broker_address}@${{env:KERBEROS_REALM}}\"; + principal=\"kafka/{controller_principal_address}@${{env:KERBEROS_REALM}}\"; }}; - ", + }; + + match role { + KafkaRole::Controller => controller_section, + KafkaRole::Broker => formatdoc! {" + bootstrap.KafkaServer {{ + com.sun.security.auth.module.Krb5LoginModule required + useKeyTab=true + storeKey=true + isInitiator=false + keyTab=\"/stackable/kerberos/keytab\" + principal=\"kafka/{bootstrap_address}@${{env:KERBEROS_REALM}}\"; + }}; + + client.KafkaServer {{ + com.sun.security.auth.module.Krb5LoginModule required + useKeyTab=true + storeKey=true + isInitiator=false + keyTab=\"/stackable/kerberos/keytab\" + principal=\"kafka/{broker_address}@${{env:KERBEROS_REALM}}\"; + }}; + + {controller_section} + ", bootstrap_address = node_address_cmd(STACKABLE_LISTENER_BOOTSTRAP_DIR), broker_address = node_address_cmd(STACKABLE_LISTENER_BROKER_DIR), }, @@ -231,15 +262,17 @@ fn jaas_config_file(is_kerberos_enabled: bool) -> String { #[cfg(test)] mod tests { use super::jaas_config_file; + use crate::crd::role::KafkaRole; #[test] fn jaas_config_file_empty_without_kerberos() { - assert_eq!(jaas_config_file(false), ""); + assert_eq!(jaas_config_file(false, &KafkaRole::Broker), ""); + assert_eq!(jaas_config_file(false, &KafkaRole::Controller), ""); } #[test] fn jaas_config_file_renders_bootstrap_and_client_sections_with_kerberos() { - let jaas = jaas_config_file(true); + let jaas = jaas_config_file(true, &KafkaRole::Broker); assert!(jaas.contains("bootstrap.KafkaServer")); assert!(jaas.contains("client.KafkaServer")); assert!(jaas.contains("Krb5LoginModule")); @@ -248,4 +281,26 @@ mod tests { assert!(jaas.contains("/stackable/listener-bootstrap")); assert!(jaas.contains("/stackable/listener-broker")); } + + #[test] + fn jaas_config_file_broker_has_controller_section_using_broker_address() { + let jaas = jaas_config_file(true, &KafkaRole::Broker); + assert!(jaas.contains("controller.KafkaServer {")); + assert!(jaas.contains( + "kafka/${file:UTF-8:/stackable/listener-broker/default-address/address}@${env:KERBEROS_REALM}" + )); + } + + #[test] + fn jaas_config_file_controller_has_controller_section_using_pod_fqdn() { + let jaas = jaas_config_file(true, &KafkaRole::Controller); + assert!(jaas.contains("controller.KafkaServer {")); + assert!(jaas.contains( + "kafka/${env:POD_NAME}.${env:ROLEGROUP_HEADLESS_SERVICE_NAME}.${env:NAMESPACE}.svc.${env:CLUSTER_DOMAIN}@${env:KERBEROS_REALM}" + )); + // Controllers have no listener-operator Listener volume, so the broker-only sections + // must not appear in their JAAS file. + assert!(!jaas.contains("bootstrap.KafkaServer")); + assert!(!jaas.contains("client.KafkaServer")); + } } From 716ee149e6803ceb4a3094b5137a9470b60aa133 Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Fri, 7 Aug 2026 15:51:11 +0200 Subject: [PATCH 07/14] feat: export KERBEROS_REALM and template jaas.properties in the KRaft controller startup command Wires Kerberos startup handling into controller_kafka_container_command, matching the broker equivalent (broker_kafka_container_commands): when Kerberos is enabled it now exports KERBEROS_REALM from krb5.conf and copies/templates jaas.properties into /tmp before kafka-storage.sh format and kafka-server-start.sh run. Behaviour is unchanged when Kerberos is disabled (ZooKeeper mode / plaintext). - controller_kafka_container_command gains a kafka_security: &ValidatedKafkaSecurity parameter (first position, matching broker_kafka_container_commands's style). - Updated the sole call site in build_controller_rolegroup_statefulset. - Added unit tests covering both the Kerberos-enabled and disabled paths. Co-Authored-By: Claude Sonnet 5 --- .../src/controller/build/command.rs | 82 +++++++++++++++++++ .../controller/build/resource/statefulset.rs | 1 + 2 files changed, 83 insertions(+) diff --git a/rust/operator-binary/src/controller/build/command.rs b/rust/operator-binary/src/controller/build/command.rs index 4805a623..7bd60f3a 100644 --- a/rust/operator-binary/src/controller/build/command.rs +++ b/rust/operator-binary/src/controller/build/command.rs @@ -157,6 +157,7 @@ wait_for_termination() "#; pub fn controller_kafka_container_command( + kafka_security: &ValidatedKafkaSecurity, controller_descriptors: Vec, product_version: &str, ) -> String { @@ -165,6 +166,7 @@ pub fn controller_kafka_container_command( {remove_vector_shutdown_file_command} prepare_signal_handlers containerdebug --output={STACKABLE_LOG_DIR}/containerdebug-state.json --loop & + {set_realm_env} POD_INDEX=$(echo \"$POD_NAME\" | grep -oE '[0-9]+$') export REPLICA_ID=$((POD_INDEX+NODE_ID_OFFSET)) @@ -173,6 +175,8 @@ pub fn controller_kafka_container_command( config-utils template /tmp/{properties_file} + {jaas_setup} + bin/kafka-storage.sh format --cluster-id \"$KAFKA_CLUSTER_ID\" --config /tmp/{properties_file} --ignore-formatted {initial_controller_command} bin/kafka-server-start.sh /tmp/{properties_file} & @@ -180,8 +184,21 @@ pub fn controller_kafka_container_command( {create_vector_shutdown_file_command} ", remove_vector_shutdown_file_command = remove_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(), + }, config_dir = STACKABLE_CONFIG_DIR, properties_file = ConfigFileName::ControllerProperties, + jaas_setup = match kafka_security.has_kerberos_enabled() { + true => formatdoc! {" + cp {config_dir}/{jaas_file} /tmp/{jaas_file} + config-utils template /tmp/{jaas_file}", + config_dir = STACKABLE_CONFIG_DIR, + jaas_file = ConfigFileName::Jaas, + }, + false => "".to_string(), + }, initial_controller_command = initial_controllers_command(&controller_descriptors, product_version), create_vector_shutdown_file_command = create_vector_shutdown_file_command(STACKABLE_LOG_DIR) } @@ -207,3 +224,68 @@ fn initial_controllers_command( ), } } + +#[cfg(test)] +mod tests { + use std::str::FromStr; + + use stackable_operator::{ + builder::meta::ObjectMetaBuilder, + crd::authentication::{core, kerberos}, + v2::types::kubernetes::SecretClassName, + }; + + use super::*; + use crate::crd::authentication::ResolvedAuthenticationClasses; + + fn kerberos_auth_class() -> core::v1alpha1::AuthenticationClass { + core::v1alpha1::AuthenticationClass { + metadata: ObjectMetaBuilder::new().name("kerberos-auth").build(), + spec: core::v1alpha1::AuthenticationClassSpec { + provider: core::v1alpha1::AuthenticationClassProvider::Kerberos( + kerberos::v1alpha1::AuthenticationProvider { + kerberos_secret_class: "kerberos-secret-class".to_string(), + }, + ), + }, + } + } + + /// Kerberos, which also requires server and internal TLS. + fn kerberos_security() -> ValidatedKafkaSecurity { + ValidatedKafkaSecurity::new( + ResolvedAuthenticationClasses::new(vec![kerberos_auth_class()]), + SecretClassName::from_str("tls").expect("tls secret class name is valid"), + Some("tls".parse().unwrap()), + None, + ) + } + + /// Plaintext: no TLS, no authentication, no OPA. + fn plaintext_security() -> ValidatedKafkaSecurity { + ValidatedKafkaSecurity::new( + ResolvedAuthenticationClasses::new(vec![]), + SecretClassName::from_str("tls").expect("tls secret class name is valid"), + None, + None, + ) + } + + #[test] + fn controller_command_exports_kerberos_realm_and_templates_jaas_when_enabled() { + let command = controller_kafka_container_command(&kerberos_security(), vec![], "4.1.1"); + assert!(command.contains("export KERBEROS_REALM=")); + assert!(command.contains(&format!( + "cp {}/jaas.properties /tmp/jaas.properties", + STACKABLE_CONFIG_DIR + ))); + assert!(command.contains("config-utils template /tmp/jaas.properties")); + } + + #[test] + fn controller_command_skips_kerberos_setup_when_disabled() { + let command = controller_kafka_container_command(&plaintext_security(), vec![], "4.1.1"); + assert!(!command.contains("KERBEROS_REALM")); + assert!(!command.contains("jaas.properties")); + } +} diff --git a/rust/operator-binary/src/controller/build/resource/statefulset.rs b/rust/operator-binary/src/controller/build/resource/statefulset.rs index d0dbf2bd..a256e6aa 100644 --- a/rust/operator-binary/src/controller/build/resource/statefulset.rs +++ b/rust/operator-binary/src/controller/build/resource/statefulset.rs @@ -498,6 +498,7 @@ pub fn build_controller_rolegroup_statefulset( "-c".to_string(), ]) .args(vec![controller_kafka_container_command( + kafka_security, validated_cluster .pod_descriptors(Some(kafka_role)) .context(BuildPodDescriptorsSnafu)?, From 6081a5757ac004cdc3826bf7d92a7da5b2450c19 Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Fri, 7 Aug 2026 15:59:12 +0200 Subject: [PATCH 08/14] fix: make controller_kafka_container_command byte-identical to pre-Kerberos output when disabled Restructure the formatdoc! template in controller_kafka_container_command so the set_realm_env and jaas_setup placeholders no longer sit next to a literal blank template line. Instead, blank-line separation is embedded in the substituted value itself, so an empty (Kerberos-disabled) substitution produces exactly the same output as before Kerberos support was added, satisfying the plan's byte-identity constraint. Add controller_command_is_byte_identical_to_pre_kerberos_output_when_disabled, which reconstructs the pre-change (commit d9942ad) template as a test-only helper and asserts real string equality against the current Kerberos-disabled output, replacing reliance on substring-only checks. Co-Authored-By: Claude Sonnet 5 --- .../src/controller/build/command.rs | 64 ++++++++++++++++--- 1 file changed, 56 insertions(+), 8 deletions(-) diff --git a/rust/operator-binary/src/controller/build/command.rs b/rust/operator-binary/src/controller/build/command.rs index 7bd60f3a..9b03c64b 100644 --- a/rust/operator-binary/src/controller/build/command.rs +++ b/rust/operator-binary/src/controller/build/command.rs @@ -167,16 +167,13 @@ pub fn controller_kafka_container_command( prepare_signal_handlers containerdebug --output={STACKABLE_LOG_DIR}/containerdebug-state.json --loop & {set_realm_env} - POD_INDEX=$(echo \"$POD_NAME\" | grep -oE '[0-9]+$') export REPLICA_ID=$((POD_INDEX+NODE_ID_OFFSET)) cp {config_dir}/{properties_file} /tmp/{properties_file} config-utils template /tmp/{properties_file} - {jaas_setup} - bin/kafka-storage.sh format --cluster-id \"$KAFKA_CLUSTER_ID\" --config /tmp/{properties_file} --ignore-formatted {initial_controller_command} bin/kafka-server-start.sh /tmp/{properties_file} & @@ -184,19 +181,23 @@ pub fn controller_kafka_container_command( {create_vector_shutdown_file_command} ", remove_vector_shutdown_file_command = remove_vector_shutdown_file_command(STACKABLE_LOG_DIR), + // When Kerberos is disabled this resolves to an empty string, so the surrounding + // template lines collapse to the same single blank line that was present before + // Kerberos support was added (byte-identical output for non-Kerberos setups). set_realm_env = match kafka_security.has_kerberos_enabled() { - true => format!("export KERBEROS_REALM=$(grep -oP 'default_realm = \\K.*' {STACKABLE_KERBEROS_KRB5_PATH})"), + true => format!("export KERBEROS_REALM=$(grep -oP 'default_realm = \\K.*' {STACKABLE_KERBEROS_KRB5_PATH})\n"), false => "".to_string(), }, config_dir = STACKABLE_CONFIG_DIR, properties_file = ConfigFileName::ControllerProperties, + // Same as `set_realm_env`: empty when Kerberos is disabled, preserving the + // pre-Kerberos-support blank-line layout. jaas_setup = match kafka_security.has_kerberos_enabled() { - true => formatdoc! {" - cp {config_dir}/{jaas_file} /tmp/{jaas_file} - config-utils template /tmp/{jaas_file}", + true => format!( + "\ncp {config_dir}/{jaas_file} /tmp/{jaas_file}\nconfig-utils template /tmp/{jaas_file}\n", config_dir = STACKABLE_CONFIG_DIR, jaas_file = ConfigFileName::Jaas, - }, + ), false => "".to_string(), }, initial_controller_command = initial_controllers_command(&controller_descriptors, product_version), @@ -288,4 +289,51 @@ mod tests { assert!(!command.contains("KERBEROS_REALM")); assert!(!command.contains("jaas.properties")); } + + /// Mirrors `controller_kafka_container_command` as it existed at commit `d9942ad` + /// (immediately before Kerberos support was added), before it took a `kafka_security` + /// parameter. Used to pin down that Kerberos-disabled output is byte-identical to the + /// pre-Kerberos-support output, per the plan's Global Constraint. + fn pre_kerberos_controller_kafka_container_command( + controller_descriptors: Vec, + product_version: &str, + ) -> String { + formatdoc! {" + {BASH_TRAP_FUNCTIONS} + {remove_vector_shutdown_file_command} + prepare_signal_handlers + containerdebug --output={STACKABLE_LOG_DIR}/containerdebug-state.json --loop & + + POD_INDEX=$(echo \"$POD_NAME\" | grep -oE '[0-9]+$') + export REPLICA_ID=$((POD_INDEX+NODE_ID_OFFSET)) + + cp {config_dir}/{properties_file} /tmp/{properties_file} + + config-utils template /tmp/{properties_file} + + bin/kafka-storage.sh format --cluster-id \"$KAFKA_CLUSTER_ID\" --config /tmp/{properties_file} --ignore-formatted {initial_controller_command} + bin/kafka-server-start.sh /tmp/{properties_file} & + + wait_for_termination $! + {create_vector_shutdown_file_command} + ", + remove_vector_shutdown_file_command = remove_vector_shutdown_file_command(STACKABLE_LOG_DIR), + config_dir = STACKABLE_CONFIG_DIR, + properties_file = ConfigFileName::ControllerProperties, + initial_controller_command = initial_controllers_command(&controller_descriptors, product_version), + create_vector_shutdown_file_command = create_vector_shutdown_file_command(STACKABLE_LOG_DIR) + } + } + + #[test] + fn controller_command_is_byte_identical_to_pre_kerberos_output_when_disabled() { + let actual = controller_kafka_container_command(&plaintext_security(), vec![], "4.1.1"); + let expected = pre_kerberos_controller_kafka_container_command(vec![], "4.1.1"); + + assert_eq!( + actual, expected, + "controller_kafka_container_command must produce byte-identical output to the \ + pre-Kerberos-support implementation when Kerberos is disabled" + ); + } } From 93d79922f1533684362c95a4d688edaf3b887cea Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Fri, 7 Aug 2026 16:09:00 +0200 Subject: [PATCH 09/14] test: add kuttl integration test for Kerberos-secured KRaft clusters Adds tests/templates/kuttl/kraft-kerberos/, combining the KRaft spec.controllers setup from smoke-kraft with the MIT KDC deployment, Kerberos AuthenticationClass/SecretClass, and kcat produce/consume job from the kerberos test case. Registers the new case as 'kraft-kerberos' in tests/test-definition.yaml alongside kafka-kraft/operations-kraft/ smoke-kraft. NOTE: untested in this environment - no reachable Kubernetes cluster was available to run kuttl against. Validated statically only (YAML/ Jinja rendering, CRD field-path cross-checks, structural comparison against smoke-kraft and kerberos). See task-8-report.md for details. Co-Authored-By: Claude Sonnet 5 --- .../kuttl/kraft-kerberos/00-assert.yaml.j2 | 10 ++ ...tor-aggregator-discovery-configmap.yaml.j2 | 9 ++ .../kuttl/kraft-kerberos/00-patch-ns.yaml.j2 | 9 ++ .../kuttl/kraft-kerberos/00-rbac.yaml.j2 | 29 ++++ .../kuttl/kraft-kerberos/01-assert.yaml.j2 | 14 ++ .../01-install-krb5-kdc.yaml.j2 | 146 ++++++++++++++++++ .../02-create-kerberos-secretclass.yaml.j2 | 72 +++++++++ .../kuttl/kraft-kerberos/20-assert.yaml | 20 +++ .../kraft-kerberos/20-install-kafka.yaml.j2 | 59 +++++++ .../kraft-kerberos/30-access-kafka.txt.j2 | 118 ++++++++++++++ .../kuttl/kraft-kerberos/30-access-kafka.yaml | 6 + .../kuttl/kraft-kerberos/30-assert.yaml | 11 ++ .../templates/kuttl/kraft-kerberos/README.md | 9 ++ tests/test-definition.yaml | 9 ++ 14 files changed, 521 insertions(+) create mode 100644 tests/templates/kuttl/kraft-kerberos/00-assert.yaml.j2 create mode 100644 tests/templates/kuttl/kraft-kerberos/00-install-vector-aggregator-discovery-configmap.yaml.j2 create mode 100644 tests/templates/kuttl/kraft-kerberos/00-patch-ns.yaml.j2 create mode 100644 tests/templates/kuttl/kraft-kerberos/00-rbac.yaml.j2 create mode 100644 tests/templates/kuttl/kraft-kerberos/01-assert.yaml.j2 create mode 100644 tests/templates/kuttl/kraft-kerberos/01-install-krb5-kdc.yaml.j2 create mode 100644 tests/templates/kuttl/kraft-kerberos/02-create-kerberos-secretclass.yaml.j2 create mode 100644 tests/templates/kuttl/kraft-kerberos/20-assert.yaml create mode 100644 tests/templates/kuttl/kraft-kerberos/20-install-kafka.yaml.j2 create mode 100644 tests/templates/kuttl/kraft-kerberos/30-access-kafka.txt.j2 create mode 100644 tests/templates/kuttl/kraft-kerberos/30-access-kafka.yaml create mode 100644 tests/templates/kuttl/kraft-kerberos/30-assert.yaml create mode 100644 tests/templates/kuttl/kraft-kerberos/README.md diff --git a/tests/templates/kuttl/kraft-kerberos/00-assert.yaml.j2 b/tests/templates/kuttl/kraft-kerberos/00-assert.yaml.j2 new file mode 100644 index 00000000..50b1d4c3 --- /dev/null +++ b/tests/templates/kuttl/kraft-kerberos/00-assert.yaml.j2 @@ -0,0 +1,10 @@ +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestAssert +{% if lookup('env', 'VECTOR_AGGREGATOR') %} +--- +apiVersion: v1 +kind: ConfigMap +metadata: + name: vector-aggregator-discovery +{% endif %} diff --git a/tests/templates/kuttl/kraft-kerberos/00-install-vector-aggregator-discovery-configmap.yaml.j2 b/tests/templates/kuttl/kraft-kerberos/00-install-vector-aggregator-discovery-configmap.yaml.j2 new file mode 100644 index 00000000..2d6a0df5 --- /dev/null +++ b/tests/templates/kuttl/kraft-kerberos/00-install-vector-aggregator-discovery-configmap.yaml.j2 @@ -0,0 +1,9 @@ +{% if lookup('env', 'VECTOR_AGGREGATOR') %} +--- +apiVersion: v1 +kind: ConfigMap +metadata: + name: vector-aggregator-discovery +data: + ADDRESS: {{ lookup('env', 'VECTOR_AGGREGATOR') }} +{% endif %} diff --git a/tests/templates/kuttl/kraft-kerberos/00-patch-ns.yaml.j2 b/tests/templates/kuttl/kraft-kerberos/00-patch-ns.yaml.j2 new file mode 100644 index 00000000..67185acf --- /dev/null +++ b/tests/templates/kuttl/kraft-kerberos/00-patch-ns.yaml.j2 @@ -0,0 +1,9 @@ +{% if test_scenario['values']['openshift'] == 'true' %} +# see https://github.com/stackabletech/issues/issues/566 +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestStep +commands: + - script: kubectl patch namespace $NAMESPACE -p '{"metadata":{"labels":{"pod-security.kubernetes.io/enforce":"privileged"}}}' + timeout: 120 +{% endif %} diff --git a/tests/templates/kuttl/kraft-kerberos/00-rbac.yaml.j2 b/tests/templates/kuttl/kraft-kerberos/00-rbac.yaml.j2 new file mode 100644 index 00000000..7ee61d23 --- /dev/null +++ b/tests/templates/kuttl/kraft-kerberos/00-rbac.yaml.j2 @@ -0,0 +1,29 @@ +--- +kind: Role +apiVersion: rbac.authorization.k8s.io/v1 +metadata: + name: test-role +rules: +{% if test_scenario['values']['openshift'] == "true" %} + - apiGroups: ["security.openshift.io"] + resources: ["securitycontextconstraints"] + resourceNames: ["privileged"] + verbs: ["use"] +{% endif %} +--- +apiVersion: v1 +kind: ServiceAccount +metadata: + name: test-sa +--- +kind: RoleBinding +apiVersion: rbac.authorization.k8s.io/v1 +metadata: + name: test-rb +subjects: + - kind: ServiceAccount + name: test-sa +roleRef: + kind: Role + name: test-role + apiGroup: rbac.authorization.k8s.io diff --git a/tests/templates/kuttl/kraft-kerberos/01-assert.yaml.j2 b/tests/templates/kuttl/kraft-kerberos/01-assert.yaml.j2 new file mode 100644 index 00000000..d34c1c63 --- /dev/null +++ b/tests/templates/kuttl/kraft-kerberos/01-assert.yaml.j2 @@ -0,0 +1,14 @@ +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestAssert +timeout: 300 +{% if test_scenario['values']['kerberos-backend'] == 'mit' %} +--- +apiVersion: apps/v1 +kind: StatefulSet +metadata: + name: krb5-kdc +status: + readyReplicas: 1 + replicas: 1 +{% endif %} diff --git a/tests/templates/kuttl/kraft-kerberos/01-install-krb5-kdc.yaml.j2 b/tests/templates/kuttl/kraft-kerberos/01-install-krb5-kdc.yaml.j2 new file mode 100644 index 00000000..69ceec81 --- /dev/null +++ b/tests/templates/kuttl/kraft-kerberos/01-install-krb5-kdc.yaml.j2 @@ -0,0 +1,146 @@ +{% if test_scenario['values']['kerberos-backend'] == 'mit' %} +apiVersion: apps/v1 +kind: StatefulSet +metadata: + name: krb5-kdc +spec: + selector: + matchLabels: + app: krb5-kdc + template: + metadata: + labels: + app: krb5-kdc + spec: + serviceAccountName: test-sa + initContainers: + - name: init + image: oci.stackable.tech/sdp/krb5:{{ test_scenario['values']['krb5'] }}-stackable0.0.0-dev + args: + - sh + - -euo + - pipefail + - -c + - | + test -e /var/kerberos/krb5kdc/principal || kdb5_util create -s -P asdf + kadmin.local get_principal -terse root/admin || kadmin.local add_principal -pw asdf root/admin + # stackable-secret-operator principal must match the keytab specified in the SecretClass + kadmin.local get_principal -terse stackable-secret-operator || kadmin.local add_principal -e aes256-cts-hmac-sha384-192:normal -pw asdf stackable-secret-operator + env: + - name: KRB5_CONFIG + value: /stackable/config/krb5.conf + volumeMounts: + - mountPath: /stackable/config + name: config + - mountPath: /var/kerberos/krb5kdc + name: data + containers: + - name: kdc + image: oci.stackable.tech/sdp/krb5:{{ test_scenario['values']['krb5'] }}-stackable0.0.0-dev + args: + - krb5kdc + - -n + env: + - name: KRB5_CONFIG + value: /stackable/config/krb5.conf + volumeMounts: + - mountPath: /stackable/config + name: config + - mountPath: /var/kerberos/krb5kdc + name: data +# Root permissions required on Openshift to bind to privileged port numbers +{% if test_scenario['values']['openshift'] == "true" %} + securityContext: + runAsUser: 0 +{% endif %} + - name: kadmind + image: oci.stackable.tech/sdp/krb5:{{ test_scenario['values']['krb5'] }}-stackable0.0.0-dev + args: + - kadmind + - -nofork + env: + - name: KRB5_CONFIG + value: /stackable/config/krb5.conf + volumeMounts: + - mountPath: /stackable/config + name: config + - mountPath: /var/kerberos/krb5kdc + name: data +# Root permissions required on Openshift to bind to privileged port numbers +{% if test_scenario['values']['openshift'] == "true" %} + securityContext: + runAsUser: 0 +{% endif %} + - name: client + image: oci.stackable.tech/sdp/krb5:{{ test_scenario['values']['krb5'] }}-stackable0.0.0-dev + tty: true + stdin: true + env: + - name: KRB5_CONFIG + value: /stackable/config/krb5.conf + volumeMounts: + - mountPath: /stackable/config + name: config + volumes: + - name: config + configMap: + name: krb5-kdc + volumeClaimTemplates: + - metadata: + name: data + spec: + accessModes: + - ReadWriteOnce + resources: + requests: + storage: 1Gi +--- +apiVersion: v1 +kind: Service +metadata: + name: krb5-kdc +spec: + selector: + app: krb5-kdc + ports: + - name: kadmin + port: 749 + - name: kdc + port: 88 + - name: kdc-udp + port: 88 + protocol: UDP +--- +apiVersion: v1 +kind: ConfigMap +metadata: + name: krb5-kdc +data: + krb5.conf: | + [logging] + default = STDERR + kdc = STDERR + admin_server = STDERR + # default = FILE:/var/log/krb5libs.log + # kdc = FILE:/var/log/krb5kdc.log + # admin_server = FILE:/vaggr/log/kadmind.log + [libdefaults] + dns_lookup_realm = false + ticket_lifetime = 24h + renew_lifetime = 7d + forwardable = true + rdns = false + default_realm = {{ test_scenario['values']['kerberos-realm'] }} + spake_preauth_groups = edwards25519 + [realms] + {{ test_scenario['values']['kerberos-realm'] }} = { + acl_file = /stackable/config/kadm5.acl + disable_encrypted_timestamp = false + } + [domain_realm] + .cluster.local = {{ test_scenario['values']['kerberos-realm'] }} + cluster.local = {{ test_scenario['values']['kerberos-realm'] }} + kadm5.acl: | + root/admin *e + stackable-secret-operator *e +{% endif %} diff --git a/tests/templates/kuttl/kraft-kerberos/02-create-kerberos-secretclass.yaml.j2 b/tests/templates/kuttl/kraft-kerberos/02-create-kerberos-secretclass.yaml.j2 new file mode 100644 index 00000000..04ae9a63 --- /dev/null +++ b/tests/templates/kuttl/kraft-kerberos/02-create-kerberos-secretclass.yaml.j2 @@ -0,0 +1,72 @@ +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestStep +commands: + - script: | + kubectl apply -n $NAMESPACE -f - < 0 %} + custom: "{{ test_scenario['values']['kafka-kraft'].split(',')[1] }}" + productVersion: "{{ test_scenario['values']['kafka-kraft'].split(',')[0] }}" +{% else %} + productVersion: "{{ test_scenario['values']['kafka-kraft'] }}" +{% endif %} + pullPolicy: IfNotPresent + clusterConfig: + # KRaft: metadata is managed by the controllers role, no ZooKeeper involved. + metadataManager: kraft + authentication: + - authenticationClass: kerberos-auth-$NAMESPACE + tls: + # Kerberos requires the use of server and internal TLS! + serverSecretClass: tls +{% if lookup('env', 'VECTOR_AGGREGATOR') %} + vectorAggregatorConfigMapName: vector-aggregator-discovery +{% endif %} + controllers: + config: + logging: + enableVectorAgent: {{ lookup('env', 'VECTOR_AGGREGATOR') | length > 0 }} + gracefulShutdownTimeout: 30s # speed up tests + roleGroups: + default: + replicas: 1 + brokers: + config: + logging: + enableVectorAgent: {{ lookup('env', 'VECTOR_AGGREGATOR') | length > 0 }} + brokerListenerClass: {{ test_scenario['values']['broker-listener-class'] }} + bootstrapListenerClass: {{ test_scenario['values']['bootstrap-listener-class'] }} + gracefulShutdownTimeout: 30s # speed up tests + roleGroups: + default: + replicas: 3 + EOF diff --git a/tests/templates/kuttl/kraft-kerberos/30-access-kafka.txt.j2 b/tests/templates/kuttl/kraft-kerberos/30-access-kafka.txt.j2 new file mode 100644 index 00000000..adeb28a4 --- /dev/null +++ b/tests/templates/kuttl/kraft-kerberos/30-access-kafka.txt.j2 @@ -0,0 +1,118 @@ +--- +apiVersion: batch/v1 +kind: Job +metadata: + name: access-kafka +spec: + template: + spec: + serviceAccountName: test-sa + containers: + - name: access-kafka + image: oci.stackable.tech/sdp/kafka:{{ test_scenario['values']['kafka'] }}-stackable0.0.0-dev + command: + - /bin/bash + - /tmp/script/script.sh + env: + - name: KRB5_CONFIG + value: /stackable/kerberos/krb5.conf + - name: KAFKA_OPTS + value: -Djava.security.krb5.conf=/stackable/kerberos/krb5.conf + - name: KAFKA + valueFrom: + configMapKeyRef: + name: test-kafka + key: KAFKA + volumeMounts: + - name: script + mountPath: /tmp/script + - mountPath: /stackable/tls-ca-cert-mount + name: tls-ca-cert-mount + - name: kerberos + mountPath: /stackable/kerberos + volumes: + - name: script + configMap: + name: access-kafka-script + - name: kerberos + ephemeral: + volumeClaimTemplate: + metadata: + annotations: + secrets.stackable.tech/class: kerberos-$NAMESPACE + secrets.stackable.tech/scope: service=access-kafka + secrets.stackable.tech/kerberos.service.names: developer + spec: + storageClassName: secrets.stackable.tech + accessModes: + - ReadWriteOnce + resources: + requests: + storage: "1" + - name: tls-ca-cert-mount + ephemeral: + volumeClaimTemplate: + metadata: + annotations: + secrets.stackable.tech/class: tls + secrets.stackable.tech/scope: pod + spec: + accessModes: + - ReadWriteOnce + resources: + requests: + storage: "1" + storageClassName: secrets.stackable.tech + volumeMode: Filesystem + securityContext: + fsGroup: 1000 + restartPolicy: OnFailure +--- +apiVersion: v1 +kind: ConfigMap +metadata: + name: access-kafka-script +data: + script.sh: | + set -euxo pipefail + + export KCAT_CONFIG=/stackable/kcat.conf + TOPIC=test-topic + CONSUMER_GROUP=test-consumer-group + + echo -e -n "\ + metadata.broker.list=$KAFKA\n\ + auto.offset.reset=beginning\n\ + security.protocol=SASL_SSL\n\ + ssl.ca.location=/stackable/tls-ca-cert-mount/ca.crt\n\ + sasl.kerberos.keytab=/stackable/kerberos/keytab\n\ + sasl.kerberos.service.name=kafka\n\ + sasl.kerberos.principal=developer/access-kafka.$NAMESPACE.svc.cluster.local@{{ test_scenario['values']['kerberos-realm'] }}\n\ + sasl.mechanism=GSSAPI\n\ + " > $KCAT_CONFIG + + cat $KCAT_CONFIG + + sent_message="Hello Stackable!" + + echo $sent_message | kcat \ + -t $TOPIC \ + -P + + echo Sent message: \"$sent_message\" + + received_message=$(kcat \ + -G $CONSUMER_GROUP \ + -o stored \ + -e \ + $TOPIC) + + echo Received message: \"$received_message\" + + if [ "$received_message" = "$sent_message" ]; then + echo "Test passed" + exit 0 + else + echo "Test failed" + exit 1 + fi diff --git a/tests/templates/kuttl/kraft-kerberos/30-access-kafka.yaml b/tests/templates/kuttl/kraft-kerberos/30-access-kafka.yaml new file mode 100644 index 00000000..eecc0f08 --- /dev/null +++ b/tests/templates/kuttl/kraft-kerberos/30-access-kafka.yaml @@ -0,0 +1,6 @@ +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestStep +commands: + # We need to replace $NAMESPACE (by KUTTL) + - script: envsubst '$NAMESPACE' < 30-access-kafka.txt | kubectl apply -n $NAMESPACE -f - diff --git a/tests/templates/kuttl/kraft-kerberos/30-assert.yaml b/tests/templates/kuttl/kraft-kerberos/30-assert.yaml new file mode 100644 index 00000000..edc6c317 --- /dev/null +++ b/tests/templates/kuttl/kraft-kerberos/30-assert.yaml @@ -0,0 +1,11 @@ +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestAssert +timeout: 600 +--- +apiVersion: batch/v1 +kind: Job +metadata: + name: access-kafka +status: + succeeded: 1 diff --git a/tests/templates/kuttl/kraft-kerberos/README.md b/tests/templates/kuttl/kraft-kerberos/README.md new file mode 100644 index 00000000..a85e47b8 --- /dev/null +++ b/tests/templates/kuttl/kraft-kerberos/README.md @@ -0,0 +1,9 @@ +# Kraft + Kerberos test + +Proves that a KRaft-mode Kafka cluster (`spec.controllers` present, no ZooKeeper) can be +secured with Kerberos authentication (`spec.clusterConfig.authentication` referencing a +Kerberos `AuthenticationClass`) end to end: controllers form a quorum, brokers join, and a +client can authenticate via GSSAPI to produce/consume a message. + +This bundles the KRaft cluster setup from `smoke-kraft` with the KDC deployment, +`SecretClass`/`AuthenticationClass` and produce/consume job from `kerberos`. diff --git a/tests/test-definition.yaml b/tests/test-definition.yaml index 3cb4633f..390257d0 100644 --- a/tests/test-definition.yaml +++ b/tests/test-definition.yaml @@ -83,6 +83,15 @@ tests: dimensions: - kafka-kraft - openshift + - name: kraft-kerberos + dimensions: + - kafka-kraft + - krb5 + - kerberos-realm + - kerberos-backend + - openshift + - broker-listener-class + - bootstrap-listener-class - name: smoke dimensions: - kafka From 7dae4d673630b6bd717013471296c1ffb030857d Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Fri, 7 Aug 2026 16:15:13 +0200 Subject: [PATCH 10/14] fix: use kafka-kraft dimension (not kafka) for access-kafka image in kraft-kerberos test The kraft-kerberos kuttl test case declares the kafka-kraft dimension in tests/test-definition.yaml, not kafka. 30-access-kafka.txt.j2 was copied verbatim from tests/templates/kuttl/kerberos/30-access-kafka.txt.j2 and still referenced test_scenario['values']['kafka'], which is Undefined for this scenario and would raise a Jinja2 UndefinedError at render time, preventing the produce/consume Job from rendering at all. Fixed the reference to use kafka-kraft, and brought the image-rendering logic in line with this same test case's 20-install-kafka.yaml.j2 (and smoke-kraft/30-install-kafka.yaml.j2), which both support the 'version[,custom-image]' convention for the kafka-kraft dimension via an if/else branch on a comma in the value. Re-ran static Jinja2/YAML validation with a stub test_scenario dict built strictly from the kraft-kerberos entry's actual dimensions (no stray kafka key this time), for both the plain-version and custom-image-comma forms of kafka-kraft. All 12 files in the test case render and parse without UndefinedError. Co-Authored-By: Claude Sonnet 5 --- tests/templates/kuttl/kraft-kerberos/30-access-kafka.txt.j2 | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/tests/templates/kuttl/kraft-kerberos/30-access-kafka.txt.j2 b/tests/templates/kuttl/kraft-kerberos/30-access-kafka.txt.j2 index adeb28a4..54aa1afa 100644 --- a/tests/templates/kuttl/kraft-kerberos/30-access-kafka.txt.j2 +++ b/tests/templates/kuttl/kraft-kerberos/30-access-kafka.txt.j2 @@ -9,7 +9,11 @@ spec: serviceAccountName: test-sa containers: - name: access-kafka - image: oci.stackable.tech/sdp/kafka:{{ test_scenario['values']['kafka'] }}-stackable0.0.0-dev +{% if test_scenario['values']['kafka-kraft'].find(",") > 0 %} + image: {{ test_scenario['values']['kafka-kraft'].split(',')[1] }} +{% else %} + image: oci.stackable.tech/sdp/kafka:{{ test_scenario['values']['kafka-kraft'] }}-stackable0.0.0-dev +{% endif %} command: - /bin/bash - /tmp/script/script.sh From 4b1d8bc6627ac659a449b22562790d530c9c42f3 Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Fri, 7 Aug 2026 16:17:33 +0200 Subject: [PATCH 11/14] docs: document Kerberos support for KRaft controllers Co-Authored-By: Claude Sonnet 5 --- CHANGELOG.md | 5 +++++ .../kafka/pages/usage-guide/kraft-controller.adoc | 11 ++++++++++- docs/modules/kafka/partials/supported-versions.adoc | 1 - 3 files changed, 15 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ef1f1c00..fd740145 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,11 @@ All notable changes to this project will be documented in this file. ## [Unreleased] +### Added + +- Kerberos authentication now works with KRaft controllers (`spec.controllers`), securing the + `CONTROLLER` listener used for broker/controller and controller/controller Raft RPC traffic ([#TBD]). + ### Changed - Internal operator refactoring: introduce a build() step in the reconciler that diff --git a/docs/modules/kafka/pages/usage-guide/kraft-controller.adoc b/docs/modules/kafka/pages/usage-guide/kraft-controller.adoc index 455188c9..20ffad21 100644 --- a/docs/modules/kafka/pages/usage-guide/kraft-controller.adoc +++ b/docs/modules/kafka/pages/usage-guide/kraft-controller.adoc @@ -91,7 +91,16 @@ KRaft mode requires major configuration changes compared to ZooKeeper: * Automatic migration from Apache ZooKeeper to KRaft is not supported. * Scaling controller replicas might lead to unstable clusters. -* Kerberos is currently not supported for KRaft in all versions. + +=== Kerberos + +Kerberos authentication is supported for KRaft clusters: enabling a Kerberos `AuthenticationClass` +secures the `CLIENT`, `INTERNAL` and `CONTROLLER` listeners alike, including controller-to-controller +and broker-to-controller Raft RPC traffic. + +NOTE: SASL/SCRAM is not supported for the controller listener by Apache Kafka itself +(https://issues.apache.org/jira/browse/KAFKA-15513[KAFKA-15513]); this does not affect Kerberos +(GSSAPI), which authenticates against the external KDC rather than Kafka-internal credential storage. == Troubleshooting diff --git a/docs/modules/kafka/partials/supported-versions.adoc b/docs/modules/kafka/partials/supported-versions.adoc index 1a57dce0..2cdf7eac 100644 --- a/docs/modules/kafka/partials/supported-versions.adoc +++ b/docs/modules/kafka/partials/supported-versions.adoc @@ -12,5 +12,4 @@ Support for clusters running in Kraft mode (which includes Apache Kafka 4.x.x) i Also there are some known issues such as: * Controller scaling is not reliable. -* Kerberos authentication is not tested yet. * Service exposition is not definitive. From ff80034914895041c307a44dcac945f7344d69b8 Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Fri, 7 Aug 2026 16:34:40 +0200 Subject: [PATCH 12/14] fix: address final whole-branch review findings for KRaft Kerberos - controller.KafkaServer JAAS section: drop isInitiator=false so the process can act as GSSAPI initiator on the CONTROLLER listener (needed since brokers connect to controllers and controllers connect to each other for Raft); add explanatory comment and a test asserting the controller section specifically lacks isInitiator=false. - kraft-kerberos kuttl test: bump controller replicas to 3 so inter-controller Kerberos-authenticated Raft traffic is actually exercised; update the corresponding assert step. - kraft-kerberos test-definition.yaml: drop the orthogonal bootstrap-listener-class dimension (already covered by the plain kerberos test case); pin it to cluster-internal in the template, following the smoke-kraft precedent. - docs: promote the Kerberos subsection out from under Known Issues into its own top-level section before Internal operator details; correct the claim that Kerberos covers the INTERNAL listener (it stays mutual TLS, unaffected by Kerberos); restore an accurate caveat in supported-versions.adoc noting the integration test is implemented/unit-tested but not yet run end-to-end against a live cluster. Co-Authored-By: Claude Sonnet 5 --- .../pages/usage-guide/kraft-controller.adoc | 21 ++++++++++--------- .../kafka/partials/supported-versions.adoc | 1 + .../controller/build/resource/config_map.rs | 17 ++++++++++++++- .../kuttl/kraft-kerberos/20-assert.yaml | 4 ++-- .../kraft-kerberos/20-install-kafka.yaml.j2 | 10 +++++++-- tests/test-definition.yaml | 1 - 6 files changed, 38 insertions(+), 16 deletions(-) diff --git a/docs/modules/kafka/pages/usage-guide/kraft-controller.adoc b/docs/modules/kafka/pages/usage-guide/kraft-controller.adoc index 20ffad21..cad71d56 100644 --- a/docs/modules/kafka/pages/usage-guide/kraft-controller.adoc +++ b/docs/modules/kafka/pages/usage-guide/kraft-controller.adoc @@ -78,6 +78,17 @@ controllers: The configuration of overrides, JVM arguments etc. is similar to the Broker and documented on the xref:concepts:overrides.adoc[concepts page]. +== Kerberos + +Kerberos authentication is supported for KRaft clusters: enabling a Kerberos `AuthenticationClass` +secures the `CLIENT`, `BOOTSTRAP` and `CONTROLLER` listeners with GSSAPI, including +controller-to-controller and broker-to-controller Raft RPC traffic on the `CONTROLLER` listener. +The `INTERNAL` inter-broker listener continues to use mutual TLS and is unaffected by Kerberos. + +NOTE: SASL/SCRAM is not supported for the controller listener by Apache Kafka itself +(https://issues.apache.org/jira/browse/KAFKA-15513[KAFKA-15513]); this does not affect Kerberos +(GSSAPI), which authenticates against the external KDC rather than Kafka-internal credential storage. + == Internal operator details KRaft mode requires major configuration changes compared to ZooKeeper: @@ -92,16 +103,6 @@ KRaft mode requires major configuration changes compared to ZooKeeper: * Automatic migration from Apache ZooKeeper to KRaft is not supported. * Scaling controller replicas might lead to unstable clusters. -=== Kerberos - -Kerberos authentication is supported for KRaft clusters: enabling a Kerberos `AuthenticationClass` -secures the `CLIENT`, `INTERNAL` and `CONTROLLER` listeners alike, including controller-to-controller -and broker-to-controller Raft RPC traffic. - -NOTE: SASL/SCRAM is not supported for the controller listener by Apache Kafka itself -(https://issues.apache.org/jira/browse/KAFKA-15513[KAFKA-15513]); this does not affect Kerberos -(GSSAPI), which authenticates against the external KDC rather than Kafka-internal credential storage. - == Troubleshooting === Cluster does not start diff --git a/docs/modules/kafka/partials/supported-versions.adoc b/docs/modules/kafka/partials/supported-versions.adoc index 2cdf7eac..4f7554b5 100644 --- a/docs/modules/kafka/partials/supported-versions.adoc +++ b/docs/modules/kafka/partials/supported-versions.adoc @@ -13,3 +13,4 @@ Also there are some known issues such as: * Controller scaling is not reliable. * Service exposition is not definitive. +* Kerberos authentication for KRaft is implemented and unit-tested, but end-to-end verification against a live cluster is still pending. diff --git a/rust/operator-binary/src/controller/build/resource/config_map.rs b/rust/operator-binary/src/controller/build/resource/config_map.rs index c35e5cf6..85e468b1 100644 --- a/rust/operator-binary/src/controller/build/resource/config_map.rs +++ b/rust/operator-binary/src/controller/build/resource/config_map.rs @@ -218,12 +218,16 @@ fn jaas_config_file(is_kerberos_enabled: bool, role: &KafkaRole) -> String { } }; + // Unlike the bootstrap/client sections below, this context is used for BOTH sides of every + // CONTROLLER-listener connection: brokers connect out to controllers, and controllers also + // connect to each other for Raft. So this is the only listener in this operator where the + // process must be able to act as a GSSAPI initiator as well as an acceptor, hence + // `isInitiator` is intentionally left at its default (`true`) here. let controller_section = formatdoc! {" controller.KafkaServer {{ com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true storeKey=true - isInitiator=false keyTab=\"/stackable/kerberos/keytab\" principal=\"kafka/{controller_principal_address}@${{env:KERBEROS_REALM}}\"; }}; @@ -302,5 +306,16 @@ mod tests { // must not appear in their JAAS file. assert!(!jaas.contains("bootstrap.KafkaServer")); assert!(!jaas.contains("client.KafkaServer")); + + // The controller.KafkaServer section must NOT set isInitiator=false: it is used both + // when brokers connect to controllers and when controllers connect to each other for + // Raft, so the process needs to be able to act as a GSSAPI initiator on this listener. + // Scope the check to the controller section itself (rather than a global absence check) + // so that a future broker-side isInitiator=false stays fine. + let controller_section_start = jaas + .find("controller.KafkaServer {") + .expect("controller.KafkaServer section must be present"); + let controller_section = &jaas[controller_section_start..]; + assert!(!controller_section.contains("isInitiator=false")); } } diff --git a/tests/templates/kuttl/kraft-kerberos/20-assert.yaml b/tests/templates/kuttl/kraft-kerberos/20-assert.yaml index abda4d5e..4e24bcec 100644 --- a/tests/templates/kuttl/kraft-kerberos/20-assert.yaml +++ b/tests/templates/kuttl/kraft-kerberos/20-assert.yaml @@ -8,8 +8,8 @@ kind: StatefulSet metadata: name: test-kafka-controller-default status: - readyReplicas: 1 - replicas: 1 + readyReplicas: 3 + replicas: 3 --- apiVersion: apps/v1 kind: StatefulSet diff --git a/tests/templates/kuttl/kraft-kerberos/20-install-kafka.yaml.j2 b/tests/templates/kuttl/kraft-kerberos/20-install-kafka.yaml.j2 index f4dbde5e..4a7052fd 100644 --- a/tests/templates/kuttl/kraft-kerberos/20-install-kafka.yaml.j2 +++ b/tests/templates/kuttl/kraft-kerberos/20-install-kafka.yaml.j2 @@ -45,13 +45,19 @@ commands: gracefulShutdownTimeout: 30s # speed up tests roleGroups: default: - replicas: 1 + # 3 controller replicas so that this test actually exercises inter-controller + # (Raft) Kerberos-authenticated traffic on the CONTROLLER listener, not just + # broker-to-controller traffic. + replicas: 3 brokers: config: logging: enableVectorAgent: {{ lookup('env', 'VECTOR_AGGREGATOR') | length > 0 }} brokerListenerClass: {{ test_scenario['values']['broker-listener-class'] }} - bootstrapListenerClass: {{ test_scenario['values']['bootstrap-listener-class'] }} + # bootstrap-listener-class is orthogonal to this test's focus on Kerberos over the + # CONTROLLER listener (that axis is already covered by the plain `kerberos` test + # case), so it is pinned here rather than parameterized as a test dimension. + bootstrapListenerClass: cluster-internal gracefulShutdownTimeout: 30s # speed up tests roleGroups: default: diff --git a/tests/test-definition.yaml b/tests/test-definition.yaml index 390257d0..086ebbfc 100644 --- a/tests/test-definition.yaml +++ b/tests/test-definition.yaml @@ -91,7 +91,6 @@ tests: - kerberos-backend - openshift - broker-listener-class - - bootstrap-listener-class - name: smoke dimensions: - kafka From ec80d67dbd9d9e480deb80df8bf0c0d531b0df52 Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Fri, 7 Aug 2026 17:11:38 +0200 Subject: [PATCH 13/14] fix: use explicit numeric offset in kraft-kerberos kcat consume step The bundled kcat's librdkafka (1.7.0) mis-detects broker feature support against Kafka >=4.0 brokers: KIP-896 dropped old low-numbered API versions, and this librdkafka version matches ApiVersions by exact version instead of range, so it wrongly reports the ListOffsets logical-offset query ("-o stored" falling back to auto.offset.reset=beginning) as unsupported even though the broker handles it fine (confluentinc/librdkafka#4948). This is unrelated to SASL/Kerberos or KRaft: broker logs show every GSSAPI handshake succeeding, including inter-controller and broker-to-controller auth over the CONTROLLER listener. Only the client-side logical-offset detection was wrong. Swapping to an explicit numeric offset ("-o 0") - valid here since the topic is freshly created and this is the only message ever produced to it - routes around the buggy code path entirely. Confirmed against a live minikube cluster with a real MIT KDC: the kuttl test (kraft-kerberos_kafka-kraft-4.2.1_..._kerberos-realm-PROD.MYCORP_...) now passes end-to-end, including 3-controller Raft quorum formation over Kerberos and real message produce/consume through the CLIENT listener. Co-Authored-By: Claude Sonnet 5 --- .../kuttl/kraft-kerberos/30-access-kafka.txt.j2 | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/tests/templates/kuttl/kraft-kerberos/30-access-kafka.txt.j2 b/tests/templates/kuttl/kraft-kerberos/30-access-kafka.txt.j2 index 54aa1afa..50a31864 100644 --- a/tests/templates/kuttl/kraft-kerberos/30-access-kafka.txt.j2 +++ b/tests/templates/kuttl/kraft-kerberos/30-access-kafka.txt.j2 @@ -105,9 +105,18 @@ data: echo Sent message: \"$sent_message\" + # Explicit numeric offset (not "-o stored"/"auto.offset.reset=beginning"): the bundled kcat's + # librdkafka (1.7.0) mis-detects broker feature support against Kafka >=4.0 brokers, which + # dropped old low-numbered API versions (KIP-896). It matches ApiVersions by exact version + # instead of range, so it wrongly reports the ListOffsets logical-offset query as unsupported + # ("Failed to query logical offset BEGINNING: Local: Required feature not supported by + # broker") even though the broker supports it fine -- see + # https://github.com/confluentinc/librdkafka/issues/4948. This is unrelated to SASL/Kerberos: + # authentication succeeds either way. The topic is freshly created and this is the only + # message ever produced to it, so offset 0 is always the message we just sent. received_message=$(kcat \ -G $CONSUMER_GROUP \ - -o stored \ + -o 0 \ -e \ $TOPIC) From df35483ba256bfcd8c4168d74ddf75e3257b4b50 Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Fri, 7 Aug 2026 17:12:06 +0200 Subject: [PATCH 14/14] docs: reflect live-cluster verification of Kerberos KRaft support The kraft-kerberos kuttl test now passes against a real minikube cluster with a real MIT KDC (Kafka 4.2.1, 3 controller replicas, GSSAPI over the CONTROLLER listener). Update the caveat that previously said end-to-end verification was pending. Co-Authored-By: Claude Sonnet 5 --- docs/modules/kafka/partials/supported-versions.adoc | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/modules/kafka/partials/supported-versions.adoc b/docs/modules/kafka/partials/supported-versions.adoc index 4f7554b5..abd30e50 100644 --- a/docs/modules/kafka/partials/supported-versions.adoc +++ b/docs/modules/kafka/partials/supported-versions.adoc @@ -13,4 +13,4 @@ Also there are some known issues such as: * Controller scaling is not reliable. * Service exposition is not definitive. -* Kerberos authentication for KRaft is implemented and unit-tested, but end-to-end verification against a live cluster is still pending. +* Kerberos authentication for KRaft is implemented, unit-tested, and has been verified end-to-end against a live cluster (Kafka 4.2.1, 3 controller replicas); it has not yet been exercised across all supported versions in CI.