From 2fec03795d781ab23e158a287427d396ab949668 Mon Sep 17 00:00:00 2001
From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com>
Date: Thu, 29 Oct 2026 18:37:09 +0100
Subject: [PATCH 01/13] feat: Add OPA support to KRaft controllers

---
 CHANGELOG.md                                  |   4 +
 .../src/controller/build/command.rs           |   3 +
 .../build/properties/broker_properties.rs     |  20 +-
 .../build/properties/controller_properties.rs |  16 +-
 .../src/controller/build/properties/mod.rs    | 111 ++++++++++-
 .../src/controller/build/security.rs          | 114 +++++++----
 .../kuttl/opa-kraft/00-patch-ns.yaml.j2       |   9 +
 .../templates/kuttl/opa-kraft/00-rbac.yaml.j2 |  29 +++
 .../templates/kuttl/opa-kraft/01-assert.yaml  |  16 ++
 .../opa-kraft/01-install-krb5-kdc.yaml.j2     | 144 ++++++++++++++
 ...vector-aggregator-discovery-configmap.yaml |  17 ++
 .../02-create-kerberos-secretclass.yaml.j2    |  37 ++++
 .../opa-kraft/09-install-secretclass.yaml.j2  |  22 +++
 .../kuttl/opa-kraft/20-assert.yaml.j2         |   8 +
 .../kuttl/opa-kraft/20-install-opa.yaml       |   7 +
 .../templates/kuttl/opa-kraft/20_opa.yaml.j2  |  77 ++++++++
 .../kuttl/opa-kraft/30-assert.yaml.j2         |  40 ++++
 .../kuttl/opa-kraft/30-install-kafka.yaml.j2  |  56 ++++++
 .../kuttl/opa-kraft/40-access-kafka.txt.j2    | 177 ++++++++++++++++++
 .../kuttl/opa-kraft/40-access-kafka.yaml      |   6 +
 .../templates/kuttl/opa-kraft/40-assert.yaml  |  11 ++
 .../vector-aggregator-values.yaml.j2          |  82 ++++++++
 tests/test-definition.yaml                    |   9 +
 23 files changed, 951 insertions(+), 64 deletions(-)
 create mode 100644 tests/templates/kuttl/opa-kraft/00-patch-ns.yaml.j2
 create mode 100644 tests/templates/kuttl/opa-kraft/00-rbac.yaml.j2
 create mode 100644 tests/templates/kuttl/opa-kraft/01-assert.yaml
 create mode 100644 tests/templates/kuttl/opa-kraft/01-install-krb5-kdc.yaml.j2
 create mode 100644 tests/templates/kuttl/opa-kraft/01-install-vector-aggregator-discovery-configmap.yaml
 create mode 100644 tests/templates/kuttl/opa-kraft/02-create-kerberos-secretclass.yaml.j2
 create mode 100644 tests/templates/kuttl/opa-kraft/09-install-secretclass.yaml.j2
 create mode 100644 tests/templates/kuttl/opa-kraft/20-assert.yaml.j2
 create mode 100644 tests/templates/kuttl/opa-kraft/20-install-opa.yaml
 create mode 100644 tests/templates/kuttl/opa-kraft/20_opa.yaml.j2
 create mode 100644 tests/templates/kuttl/opa-kraft/30-assert.yaml.j2
 create mode 100644 tests/templates/kuttl/opa-kraft/30-install-kafka.yaml.j2
 create mode 100644 tests/templates/kuttl/opa-kraft/40-access-kafka.txt.j2
 create mode 100644 tests/templates/kuttl/opa-kraft/40-access-kafka.yaml
 create mode 100644 tests/templates/kuttl/opa-kraft/40-assert.yaml
 create mode 100644 tests/templates/kuttl/opa-kraft/vector-aggregator-values.yaml.j2

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 4fd40a4b0..5c29a8dbc 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -13,6 +13,9 @@ All notable changes to this project will be documented in this file.
 - Support for Kafka 4.3.1 ([#1022]).
 - Add `/ready` endpoint to the operator Deployment, which reports the CRD installation status ([#1029]).
 - Brokers now have a default affinity to the OPA Pods when OPA authorization is configured ([#1035]).
+- The OPA authorizer is now also configured on KRaft controllers. Previously, requests that brokers
+  forward to the controllers (e.g. creating or deleting topics) were not authorized at all.
+  The Kafka nodes are super users, so that traffic between them does not hit OPA ([#XXX]).
 
 ### Changed
 
@@ -99,6 +102,7 @@ All notable changes to this project will be documented in this file.
 [#1046]: https://github.com/stackabletech/kafka-operator/pull/1046
 [#1047]: https://github.com/stackabletech/kafka-operator/pull/1047
 [#1049]: https://github.com/stackabletech/kafka-operator/pull/1049
+[#XXX]: https://github.com/stackabletech/kafka-operator/issues/XXX
 
 ## [26.7.0] - 2026-07-21
 
diff --git a/rust/operator-binary/src/controller/build/command.rs b/rust/operator-binary/src/controller/build/command.rs
index ebe59764d..47fa68b10 100644
--- a/rust/operator-binary/src/controller/build/command.rs
+++ b/rust/operator-binary/src/controller/build/command.rs
@@ -157,6 +157,8 @@ pub fn controller_kafka_container_command(
         containerdebug --output={STACKABLE_LOG_DIR}/containerdebug-state.json --loop &
         {set_realm_env}
 
+        {import_opa_tls_cert}
+
         {derive_pod_index}
         {export_replica_id}
 
@@ -180,6 +182,7 @@ pub fn controller_kafka_container_command(
         remove_vector_shutdown_file_command = remove_vector_shutdown_file_command(STACKABLE_LOG_DIR),
         // Mirrors `broker_kafka_container_commands`: empty when Kerberos is disabled.
         set_realm_env = export_kerberos_realm_command(kafka_security).unwrap_or_default(),
+        import_opa_tls_cert = copy_opa_tls_cert_command(kafka_security),
         derive_pod_index = DERIVE_POD_INDEX,
         export_replica_id = EXPORT_REPLICA_ID,
         config_dir = STACKABLE_CONFIG_DIR,
diff --git a/rust/operator-binary/src/controller/build/properties/broker_properties.rs b/rust/operator-binary/src/controller/build/properties/broker_properties.rs
index ee76e0a78..9dee8b335 100644
--- a/rust/operator-binary/src/controller/build/properties/broker_properties.rs
+++ b/rust/operator-binary/src/controller/build/properties/broker_properties.rs
@@ -1,6 +1,6 @@
 use std::collections::BTreeMap;
 
-use super::kraft_controllers;
+use super::{kraft_controllers, opa_authorizer_properties};
 use crate::{
     controller::{
         ValidatedClusterConfig,
@@ -95,23 +95,7 @@ pub fn build(
         }
     }
 
-    // Enable OPA authorization
-    if let Some(opa_connect_string) = cluster_config.opa_connect() {
-        result.extend([
-            (
-                "authorizer.class.name".to_string(),
-                "org.openpolicyagent.kafka.OpaAuthorizer".to_string(),
-            ),
-            (
-                "opa.authorizer.metrics.enabled".to_string(),
-                "true".to_string(),
-            ),
-            (
-                "opa.authorizer.url".to_string(),
-                opa_connect_string.to_string(),
-            ),
-        ]);
-    }
+    result.extend(opa_authorizer_properties(cluster_config, pod_descriptors));
 
     result.extend(broker_config_settings(&cluster_config.kafka_security));
     result.extend(graceful_shutdown_config_properties());
diff --git a/rust/operator-binary/src/controller/build/properties/controller_properties.rs b/rust/operator-binary/src/controller/build/properties/controller_properties.rs
index 6804653f2..128a7ef91 100644
--- a/rust/operator-binary/src/controller/build/properties/controller_properties.rs
+++ b/rust/operator-binary/src/controller/build/properties/controller_properties.rs
@@ -1,12 +1,12 @@
 use std::collections::BTreeMap;
 
-use super::kraft_controllers;
+use super::{kraft_controllers, opa_authorizer_properties};
 use crate::{
     controller::{
         ValidatedClusterConfig,
         build::{
             graceful_shutdown::graceful_shutdown_config_properties,
-            security::controller_config_settings,
+            security::{SSL_CLIENT_AUTH_REQUIRED, controller_config_settings},
         },
     },
     crd::{
@@ -73,6 +73,18 @@ pub fn build(
         );
     }
 
+    result.extend(opa_authorizer_properties(cluster_config, pod_descriptors));
+    if cluster_config.opa_connect().is_some()
+        && !cluster_config.kafka_security.has_kerberos_enabled()
+    {
+        // The nodes are identified as super users by the subject of their internal certificate.
+        // Without client authentication, the principal on the CONTROLLER listener is ANONYMOUS.
+        // With Kerberos, the principal is derived from the SASL authentication instead.
+        result.insert(
+            KafkaListenerName::Controller.listener_ssl_client_auth(),
+            SSL_CLIENT_AUTH_REQUIRED.to_string(),
+        );
+    }
     result.extend(controller_config_settings(&cluster_config.kafka_security));
     result.extend(graceful_shutdown_config_properties());
     result.extend(overrides);
diff --git a/rust/operator-binary/src/controller/build/properties/mod.rs b/rust/operator-binary/src/controller/build/properties/mod.rs
index e751613aa..7e85c5e3f 100644
--- a/rust/operator-binary/src/controller/build/properties/mod.rs
+++ b/rust/operator-binary/src/controller/build/properties/mod.rs
@@ -6,11 +6,14 @@ pub mod listener;
 pub mod product_logging;
 pub mod security_properties;
 
-use std::collections::BTreeSet;
+use std::collections::{BTreeMap, BTreeSet};
 
-use crate::crd::{
-    KafkaPodDescriptor,
-    role::{AnyConfig, KafkaRole},
+use crate::{
+    controller::ValidatedClusterConfig,
+    crd::{
+        KafkaPodDescriptor,
+        role::{AnyConfig, KafkaRole},
+    },
 };
 
 /// The names of the config files assembled into the rolegroup `ConfigMap`.
@@ -117,6 +120,84 @@ pub(crate) fn kraft_controllers(
         .collect()
 }
 
+/// The OPA authorizer properties shared by brokers and controllers.
+///
+/// In KRaft mode, brokers forward requests such as `CreateTopics` to the controllers without
+/// authorizing them, so the controllers must run the authorizer too. Empty if OPA is not
+/// configured.
+///
+/// The Kafka nodes themselves are super users, so that traffic between them does not
+/// hit OPA. Forwarded requests are still authorized against the original client principal.
+pub(crate) fn opa_authorizer_properties(
+    cluster_config: &ValidatedClusterConfig,
+    pod_descriptors: &[KafkaPodDescriptor],
+) -> BTreeMap<String, String> {
+    let Some(opa_connect_string) = cluster_config.opa_connect() else {
+        return BTreeMap::new();
+    };
+
+    BTreeMap::from([
+        (
+            "authorizer.class.name".to_string(),
+            "org.openpolicyagent.kafka.OpaAuthorizer".to_string(),
+        ),
+        (
+            "opa.authorizer.metrics.enabled".to_string(),
+            "true".to_string(),
+        ),
+        (
+            "opa.authorizer.url".to_string(),
+            opa_connect_string.to_string(),
+        ),
+        (
+            "super.users".to_string(),
+            super_users(
+                pod_descriptors,
+                cluster_config.kafka_security.has_kerberos_enabled(),
+            ),
+        ),
+    ])
+}
+
+/// The `super.users` value: the principals of all Kafka nodes.
+///
+/// Internal traffic is authenticated with the pod certificates of the internal `SecretClass`
+/// (mTLS), whose subject DN is `CN=generated certificate for pod,DC=<pod>,DC=<service>,
+/// DC=<namespace>,DC=svc,DC=<cluster domain parts>`, see the secret-operator.
+/// With Kerberos, the node principals are mapped to the short name of the service.
+fn super_users(pod_descriptors: &[KafkaPodDescriptor], kerberos_enabled: bool) -> String {
+    let mut principals: BTreeSet<String> = pod_descriptors
+        .iter()
+        .map(|desc| {
+            // The certificate encodes `CN`, then the `DC`s from the pod up to the cluster domain
+            // suffix. Java prints the subject DN in RFC 2253 order, which is reversed.
+            let cluster_domain = desc.cluster_domain.to_string();
+            let domain_components = [
+                format!("{}-{}", desc.role_group_statefulset_name, desc.replica),
+                desc.role_group_service_name.to_string(),
+                desc.namespace.to_string(),
+                "svc".to_string(),
+            ]
+            .into_iter()
+            .chain(cluster_domain.split('.').map(str::to_string));
+            let mut rdns: Vec<String> = vec!["CN=generated certificate for pod".to_string()];
+            rdns.extend(domain_components.map(|dc| format!("DC={dc}")));
+            rdns.reverse();
+            format!("User:{}", rdns.join(","))
+        })
+        .collect();
+
+    if kerberos_enabled {
+        principals.extend(
+            [KafkaRole::Broker, KafkaRole::Controller]
+                .iter()
+                .map(|role| format!("User:{}", role.kerberos_service_name())),
+        );
+    }
+
+    principals.into_iter().collect::<Vec<_>>().join(";")
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -162,6 +243,28 @@ mod tests {
         }
     }
 
+    #[test]
+    fn super_users_contains_the_pod_certificate_subject_of_every_node() {
+        let pod_descriptors = vec![
+            pod_descriptor(KafkaRole::Controller, 0, 9093),
+            pod_descriptor(KafkaRole::Broker, 1, 9092),
+        ];
+
+        assert_eq!(
+            super_users(&pod_descriptors, false),
+            "User:DC=local,DC=cluster,DC=svc,DC=default,DC=kafka-controller-default-headless,DC=kafka-controller-default-0,CN=generated certificate for pod;\
+             User:DC=local,DC=cluster,DC=svc,DC=default,DC=kafka-controller-default-headless,DC=kafka-controller-default-1,CN=generated certificate for pod"
+        );
+    }
+
+    #[test]
+    fn super_users_contains_the_service_name_with_kerberos() {
+        let pod_descriptors = vec![pod_descriptor(KafkaRole::Controller, 0, 9093)];
+
+        assert!(super_users(&pod_descriptors, true).ends_with(";User:kafka"));
+        assert!(!super_users(&pod_descriptors, false).contains("User:kafka"));
+    }
+
     #[test]
     fn kraft_controllers_lists_individual_pod_fqdns_under_kerberos() {
         let pod_descriptors = vec![
diff --git a/rust/operator-binary/src/controller/build/security.rs b/rust/operator-binary/src/controller/build/security.rs
index 0906bf7ef..cfaf4fb5d 100644
--- a/rust/operator-binary/src/controller/build/security.rs
+++ b/rust/operator-binary/src/controller/build/security.rs
@@ -44,7 +44,7 @@ const OPA_TLS_MOUNT_PATH: &str = "/stackable/tls-opa";
 constant!(OPA_TLS_VOLUME_NAME: VolumeName = "tls-opa");
 const SSL_STORE_PASSWORD: &str = "";
 const SSL_STORE_TYPE_PKCS12: &str = "PKCS12";
-const SSL_CLIENT_AUTH_REQUIRED: &str = "required";
+pub(super) const SSL_CLIENT_AUTH_REQUIRED: &str = "required";
 const SASL_MECHANISM_GSSAPI: &str = "GSSAPI";
 const STACKABLE_KCAT_BINARY: &str = "/stackable/kcat";
 const PROPERTY_SECURITY_PROTOCOL: &str = "security.protocol";
@@ -263,6 +263,39 @@ pub fn controller_admin_client_properties(
     properties
 }
 
+/// Adds the volume and volume mount holding the CA of the OPA TLS `SecretClass`, if configured.
+///
+/// The CA is imported into the internal truststore on startup, see [`copy_opa_tls_cert_command`].
+fn add_opa_tls_volume_and_volume_mount(
+    security: &ValidatedKafkaSecurity,
+    pod_builder: &mut PodBuilder,
+    cb_kafka: &mut ContainerBuilder,
+) -> Result<(), Error> {
+    if let Some(secret_class) = security.opa_secret_class() {
+        cb_kafka
+            .add_volume_mount(&*OPA_TLS_VOLUME_NAME, OPA_TLS_MOUNT_PATH)
+            .expect("The mount paths are statically defined and there should be no duplicates.");
+
+        pod_builder
+            .add_volume(
+                VolumeBuilder::new(&*OPA_TLS_VOLUME_NAME)
+                    .ephemeral(
+                        SecretOperatorVolumeSourceBuilder::new(
+                            secret_class,
+                            // Only the truststore is required to connect to OPA.
+                            SecretClassVolumeProvisionParts::Public,
+                        )
+                        .build()
+                        .context(OpaTlsCertSecretClassVolumeBuildSnafu)?,
+                    )
+                    .build(),
+            )
+            .context(AddVolumeSnafu)?;
+    }
+
+    Ok(())
+}
+
 /// Adds required volumes and volume mounts to the broker pod and container builders
 /// depending on the tls and authentication settings.
 ///
@@ -320,27 +353,7 @@ pub fn add_broker_volume_and_volume_mounts(
         )
         .expect("The mount paths are statically defined and there should be no duplicates.");
 
-    if let Some(secret_class) = security.opa_secret_class() {
-        cb_kafka
-            .add_volume_mount(&*OPA_TLS_VOLUME_NAME, OPA_TLS_MOUNT_PATH)
-            .expect("The mount paths are statically defined and there should be no duplicates.");
-
-        pod_builder
-            .add_volume(
-                VolumeBuilder::new(&*OPA_TLS_VOLUME_NAME)
-                    .ephemeral(
-                        SecretOperatorVolumeSourceBuilder::new(
-                            secret_class,
-                            // Only the truststore is required to connect to OPA.
-                            SecretClassVolumeProvisionParts::Public,
-                        )
-                        .build()
-                        .context(OpaTlsCertSecretClassVolumeBuildSnafu)?,
-                    )
-                    .build(),
-            )
-            .context(AddVolumeSnafu)?;
-    }
+    add_opa_tls_volume_and_volume_mount(security, pod_builder, cb_kafka)?;
 
     Ok(())
 }
@@ -385,6 +398,8 @@ pub fn add_controller_volume_and_volume_mounts(
         )
         .expect("The mount paths are statically defined and there should be no duplicates.");
 
+    add_opa_tls_volume_and_volume_mount(security, pod_builder, cb_kafka)?;
+
     Ok(())
 }
 
@@ -456,6 +471,32 @@ fn push_client_ssl_truststore(props: &mut Vec<(String, Option<String>)>, dir: &s
     ));
 }
 
+/// Inserts the `opa.authorizer.truststore.*` settings if OPA is configured with TLS.
+///
+/// The OPA CA is imported into the internal truststore, see [`copy_opa_tls_cert_command`].
+fn insert_opa_truststore_settings(
+    config: &mut BTreeMap<String, String>,
+    security: &ValidatedKafkaSecurity,
+) {
+    if security.opa_secret_class().is_some() {
+        config.insert(
+            "opa.authorizer.truststore.path".to_string(),
+            format!(
+                "{}/{}",
+                STACKABLE_TLS_KAFKA_INTERNAL_DIR, TRUSTSTORE_P12_FILE_NAME
+            ),
+        );
+        config.insert(
+            "opa.authorizer.truststore.password".to_string(),
+            SSL_STORE_PASSWORD.to_string(),
+        );
+        config.insert(
+            "opa.authorizer.truststore.type".to_string(),
+            SSL_STORE_TYPE_PKCS12.to_string(),
+        );
+    }
+}
+
 /// Returns required Kafka configuration settings for the `broker.properties` file
 /// depending on the tls and authentication settings.
 pub fn broker_config_settings(security: &ValidatedKafkaSecurity) -> BTreeMap<String, String> {
@@ -526,24 +567,7 @@ pub fn broker_config_settings(security: &ValidatedKafkaSecurity) -> BTreeMap<Str
         SSL_CLIENT_AUTH_REQUIRED.to_string(),
     );
 
-    //OPA Tls
-    if security.opa_secret_class().is_some() {
-        config.insert(
-            "opa.authorizer.truststore.path".to_string(),
-            format!(
-                "{}/{}",
-                STACKABLE_TLS_KAFKA_INTERNAL_DIR, TRUSTSTORE_P12_FILE_NAME
-            ),
-        );
-        config.insert(
-            "opa.authorizer.truststore.password".to_string(),
-            SSL_STORE_PASSWORD.to_string(),
-        );
-        config.insert(
-            "opa.authorizer.truststore.type".to_string(),
-            SSL_STORE_TYPE_PKCS12.to_string(),
-        );
-    }
+    insert_opa_truststore_settings(&mut config, security);
 
     // common
     config.insert(
@@ -583,6 +607,8 @@ pub fn controller_config_settings(security: &ValidatedKafkaSecurity) -> BTreeMap
         );
     }
 
+    insert_opa_truststore_settings(&mut config, security);
+
     // Kerberos
     if security.has_kerberos_enabled() {
         config.insert(
@@ -1159,6 +1185,14 @@ pub(crate) mod tests {
 
     // ---- controller_config_settings ----
 
+    #[test]
+    fn controller_config_opa_adds_authorizer_truststore() {
+        let config = controller_config_settings(&opa());
+        assert!(config.contains_key("opa.authorizer.truststore.path"));
+        assert!(config.contains_key("opa.authorizer.truststore.password"));
+        assert!(config.contains_key("opa.authorizer.truststore.type"));
+    }
+
     #[test]
     fn controller_config_plaintext_has_internal_tls() {
         // Internal TLS is mandatory, so the controller config is non-empty even for a plaintext
diff --git a/tests/templates/kuttl/opa-kraft/00-patch-ns.yaml.j2 b/tests/templates/kuttl/opa-kraft/00-patch-ns.yaml.j2
new file mode 100644
index 000000000..67185acfd
--- /dev/null
+++ b/tests/templates/kuttl/opa-kraft/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/opa-kraft/00-rbac.yaml.j2 b/tests/templates/kuttl/opa-kraft/00-rbac.yaml.j2
new file mode 100644
index 000000000..7ee61d23e
--- /dev/null
+++ b/tests/templates/kuttl/opa-kraft/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/opa-kraft/01-assert.yaml b/tests/templates/kuttl/opa-kraft/01-assert.yaml
new file mode 100644
index 000000000..615b16ab1
--- /dev/null
+++ b/tests/templates/kuttl/opa-kraft/01-assert.yaml
@@ -0,0 +1,16 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestAssert
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: vector-aggregator-discovery
+---
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+  name: krb5-kdc
+status:
+  readyReplicas: 1
+  replicas: 1
diff --git a/tests/templates/kuttl/opa-kraft/01-install-krb5-kdc.yaml.j2 b/tests/templates/kuttl/opa-kraft/01-install-krb5-kdc.yaml.j2
new file mode 100644
index 000000000..b77a5fc20
--- /dev/null
+++ b/tests/templates/kuttl/opa-kraft/01-install-krb5-kdc.yaml.j2
@@ -0,0 +1,144 @@
+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 = CLUSTER.LOCAL
+    spake_preauth_groups = edwards25519
+    [realms]
+    CLUSTER.LOCAL = {
+     acl_file = /stackable/config/kadm5.acl
+     disable_encrypted_timestamp = false
+    }
+    [domain_realm]
+    .cluster.local = CLUSTER.LOCAL
+    cluster.local = CLUSTER.LOCAL
+  kadm5.acl: |
+    root/admin *e
+    stackable-secret-operator *e
diff --git a/tests/templates/kuttl/opa-kraft/01-install-vector-aggregator-discovery-configmap.yaml b/tests/templates/kuttl/opa-kraft/01-install-vector-aggregator-discovery-configmap.yaml
new file mode 100644
index 000000000..91752a602
--- /dev/null
+++ b/tests/templates/kuttl/opa-kraft/01-install-vector-aggregator-discovery-configmap.yaml
@@ -0,0 +1,17 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestStep
+commands:
+  - script: >-
+      helm install vector-aggregator vector
+      --namespace $NAMESPACE
+      --version 0.58.0 `# app version 0.58.0`
+      --repo https://helm.vector.dev
+      --values vector-aggregator-values.yaml
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: vector-aggregator-discovery
+data:
+  ADDRESS: http://vector-aggregator:6123
diff --git a/tests/templates/kuttl/opa-kraft/02-create-kerberos-secretclass.yaml.j2 b/tests/templates/kuttl/opa-kraft/02-create-kerberos-secretclass.yaml.j2
new file mode 100644
index 000000000..0d443b614
--- /dev/null
+++ b/tests/templates/kuttl/opa-kraft/02-create-kerberos-secretclass.yaml.j2
@@ -0,0 +1,37 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestStep
+commands:
+  - script: |
+      kubectl apply -n $NAMESPACE -f - <<EOF
+      ---
+      apiVersion: secrets.stackable.tech/v1alpha1
+      kind: SecretClass
+      metadata:
+        name: kerberos-$NAMESPACE
+      spec:
+        backend:
+          kerberosKeytab:
+            realmName: CLUSTER.LOCAL
+            kdc: krb5-kdc.$NAMESPACE.svc.cluster.local
+            admin:
+              mit:
+                kadminServer: krb5-kdc.$NAMESPACE.svc.cluster.local
+            adminKeytabSecret:
+              namespace: $NAMESPACE
+              name: secret-operator-keytab
+            adminPrincipal: stackable-secret-operator
+      EOF
+---
+apiVersion: v1
+kind: Secret
+metadata:
+  name: secret-operator-keytab
+data:
+  # To create keytab. When prompted enter password asdf
+  # cat | ktutil << 'EOF'
+  # list
+  # add_entry -password -p stackable-secret-operator@CLUSTER.LOCAL -k 1 -e aes256-cts-hmac-sha384-192
+  # wkt /tmp/keytab
+  # EOF
+  keytab: BQIAAABdAAEADUNMVVNURVIuTE9DQUwAGXN0YWNrYWJsZS1zZWNyZXQtb3BlcmF0b3IAAAABZAYWIgEAFAAgm8MCZ8B//XF1tH92GciD6/usWUNAmBTZnZQxLua2TkgAAAAB
diff --git a/tests/templates/kuttl/opa-kraft/09-install-secretclass.yaml.j2 b/tests/templates/kuttl/opa-kraft/09-install-secretclass.yaml.j2
new file mode 100644
index 000000000..8a09c4332
--- /dev/null
+++ b/tests/templates/kuttl/opa-kraft/09-install-secretclass.yaml.j2
@@ -0,0 +1,22 @@
+{% if test_scenario['values']['use-opa-tls'] == "true" %}
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestStep
+commands:
+  - script: |
+      kubectl apply -n $NAMESPACE -f - << EOF
+      ---
+      apiVersion: secrets.stackable.tech/v1alpha2
+      kind: SecretClass
+      metadata:
+        name: opa-tls-$NAMESPACE
+      spec:
+        backend:
+          autoTls:
+            ca:
+              autoGenerate: true
+              secret:
+                name: opa-tls-ca
+                namespace: $NAMESPACE
+      EOF
+{% endif %}
diff --git a/tests/templates/kuttl/opa-kraft/20-assert.yaml.j2 b/tests/templates/kuttl/opa-kraft/20-assert.yaml.j2
new file mode 100644
index 000000000..819abc51a
--- /dev/null
+++ b/tests/templates/kuttl/opa-kraft/20-assert.yaml.j2
@@ -0,0 +1,8 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestAssert
+metadata:
+  name: install-opa
+timeout: 300
+commands:
+  - script: kubectl -n $NAMESPACE wait --for=condition=available opaclusters.opa.stackable.tech/test-opa --timeout 301s
diff --git a/tests/templates/kuttl/opa-kraft/20-install-opa.yaml b/tests/templates/kuttl/opa-kraft/20-install-opa.yaml
new file mode 100644
index 000000000..8adb9177a
--- /dev/null
+++ b/tests/templates/kuttl/opa-kraft/20-install-opa.yaml
@@ -0,0 +1,7 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestStep
+commands:
+  - script: >
+      envsubst '$NAMESPACE' < 20_opa.yaml |
+      kubectl apply -n $NAMESPACE -f -
diff --git a/tests/templates/kuttl/opa-kraft/20_opa.yaml.j2 b/tests/templates/kuttl/opa-kraft/20_opa.yaml.j2
new file mode 100644
index 000000000..a98e20aac
--- /dev/null
+++ b/tests/templates/kuttl/opa-kraft/20_opa.yaml.j2
@@ -0,0 +1,77 @@
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: test-rego-kafka
+  labels:
+    opa.stackable.tech/bundle: "true"
+data:
+  kafka.rego: |
+    package authz
+
+    default allow := false
+
+    # The Kafka nodes are super users and do not hit OPA for their own traffic,
+    # so only the developer needs a rule.
+    allow if {
+      input.requestContext.principal.name == "developer"
+      resource_is_allowed
+    }
+
+    # The (idempotent) producer needs this cluster-level permission.
+    allow if {
+      input.requestContext.principal.name == "developer"
+      input.action.resourcePattern.resourceType == "CLUSTER"
+      input.action.operation == "IDEMPOTENT_WRITE"
+    }
+
+    # only allow access to a specific topic
+    resource_is_allowed if {
+      input.action.resourcePattern.resourceType == "TOPIC"
+      input.action.resourcePattern.name == "test-topic"
+      action_is_allowed
+    }
+
+    resource_is_allowed if {
+      input.action.resourcePattern.resourceType == "GROUP"
+      input.action.operation in ["DESCRIBE", "READ"]
+    }
+
+    action_is_allowed if {
+      input.action.operation in ["CREATE", "DESCRIBE", "READ", "WRITE"]
+    }
+---
+apiVersion: opa.stackable.tech/v1alpha1
+kind: OpaCluster
+metadata:
+  name: test-opa
+spec:
+  image:
+{% if test_scenario['values']['opa-latest'].find(",") > 0 %}
+    custom: "{{ test_scenario['values']['opa-latest'].split(',')[1] }}"
+    productVersion: "{{ test_scenario['values']['opa-latest'].split(',')[0] }}"
+{% else %}
+    productVersion: "{{ test_scenario['values']['opa-latest'] }}"
+{% endif %}
+    pullPolicy: IfNotPresent
+  clusterConfig:
+{% if test_scenario['values']['use-opa-tls'] == "true" %}
+    tls:
+      serverSecretClass: opa-tls-$NAMESPACE
+{% endif %}
+    vectorAggregatorConfigMapName: vector-aggregator-discovery
+  servers:
+    config:
+      logging:
+        enableVectorAgent: true
+        containers:
+          opa:
+            console:
+              level: INFO
+            file:
+              level: INFO
+            loggers:
+              decision:
+                level: INFO
+    roleGroups:
+      default: {}
diff --git a/tests/templates/kuttl/opa-kraft/30-assert.yaml.j2 b/tests/templates/kuttl/opa-kraft/30-assert.yaml.j2
new file mode 100644
index 000000000..55d42ab0d
--- /dev/null
+++ b/tests/templates/kuttl/opa-kraft/30-assert.yaml.j2
@@ -0,0 +1,40 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestAssert
+timeout: 300
+---
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+  name: test-kafka-controller-default
+status:
+  readyReplicas: 3
+  replicas: 3
+---
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+  name: test-kafka-broker-default
+status:
+  readyReplicas: 3
+  replicas: 3
+---
+apiVersion: v1
+kind: PersistentVolumeClaim
+metadata:
+  name: log-dirs-test-kafka-broker-default-0
+spec:
+  resources:
+    requests:
+      storage: 2Gi
+status:
+  phase: Bound
+---
+apiVersion: policy/v1
+kind: PodDisruptionBudget
+metadata:
+  name: test-kafka-broker
+status:
+  expectedPods: 3
+  currentHealthy: 3
+  disruptionsAllowed: 1
diff --git a/tests/templates/kuttl/opa-kraft/30-install-kafka.yaml.j2 b/tests/templates/kuttl/opa-kraft/30-install-kafka.yaml.j2
new file mode 100644
index 000000000..b897a9216
--- /dev/null
+++ b/tests/templates/kuttl/opa-kraft/30-install-kafka.yaml.j2
@@ -0,0 +1,56 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestStep
+commands:
+  - script: |
+      kubectl apply -n $NAMESPACE -f - <<EOF
+      ---
+      apiVersion: authentication.stackable.tech/v1alpha1
+      kind: AuthenticationClass
+      metadata:
+        name: kerberos-auth-$NAMESPACE
+      spec:
+        provider:
+          kerberos:
+            kerberosSecretClass: kerberos-$NAMESPACE
+      ---
+      apiVersion: kafka.stackable.tech/v1alpha1
+      kind: KafkaCluster
+      metadata:
+        name: test-kafka
+      spec:
+        image:
+{% if test_scenario['values']['kafka-kraft'].find(",") > 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
+          authorization:
+            opa:
+              configMapName: test-opa
+              package: authz
+          tls:
+            serverSecretClass: tls
+          vectorAggregatorConfigMapName: vector-aggregator-discovery
+        controllers:
+          config:
+            logging:
+              enableVectorAgent: true
+          roleGroups:
+            default:
+              replicas: 3
+        brokers:
+          config:
+            logging:
+              enableVectorAgent: true
+          roleGroups:
+            default:
+              replicas: 3
+      EOF
diff --git a/tests/templates/kuttl/opa-kraft/40-access-kafka.txt.j2 b/tests/templates/kuttl/opa-kraft/40-access-kafka.txt.j2
new file mode 100644
index 000000000..93bf47fc9
--- /dev/null
+++ b/tests/templates/kuttl/opa-kraft/40-access-kafka.txt.j2
@@ -0,0 +1,177 @@
+---
+apiVersion: batch/v1
+kind: Job
+metadata:
+  name: access-kafka
+spec:
+  template:
+    spec:
+      serviceAccountName: test-sa
+      containers:
+        - name: access-kafka
+{% 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
+          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
+
+    TOPIC=test-topic
+    DISALLOWED_TOPIC=disallowed-topic
+    CONSUMER_GROUP=test-consumer-group
+    ADMIN_CONFIG=/stackable/admin.properties
+
+    # In KRaft mode, the brokers forward topic management requests to the controllers, which
+    # have to authorize them with OPA. Test this with the Kafka admin client.
+    cat > $ADMIN_CONFIG <<EOF
+    security.protocol=SASL_SSL
+    sasl.mechanism=GSSAPI
+    sasl.kerberos.service.name=kafka
+    ssl.truststore.type=PEM
+    ssl.truststore.location=/stackable/tls-ca-cert-mount/ca.crt
+    sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true storeKey=true keyTab="/stackable/kerberos/keytab" principal="developer/access-kafka.$NAMESPACE.svc.cluster.local@CLUSTER.LOCAL";
+    EOF
+
+    # Temporarily disable exit-on-error to be able to assert on failures
+    set +e
+
+    echo "Creating a disallowed topic must fail"
+    output=$(/stackable/kafka/bin/kafka-topics.sh --bootstrap-server $KAFKA --command-config $ADMIN_CONFIG --create --topic $DISALLOWED_TOPIC 2>&1)
+    exit_code=$?
+    echo "$output" | tail -5
+    if [ $exit_code -eq 0 ] || ! echo "$output" | grep -qi "authorization failed"; then
+      echo "Test failed: creating $DISALLOWED_TOPIC should have been denied"
+      exit 1
+    fi
+
+    set -e
+
+    echo "Creating the allowed topic must succeed"
+    /stackable/kafka/bin/kafka-topics.sh --bootstrap-server $KAFKA --command-config $ADMIN_CONFIG --create --if-not-exists --topic $TOPIC
+
+    set +e
+
+    echo "Deleting the allowed topic must fail, as only CREATE, DESCRIBE, READ and WRITE are allowed"
+    output=$(/stackable/kafka/bin/kafka-topics.sh --bootstrap-server $KAFKA --command-config $ADMIN_CONFIG --delete --topic $TOPIC 2>&1)
+    exit_code=$?
+    echo "$output" | tail -5
+    if [ $exit_code -eq 0 ] || ! echo "$output" | grep -qi "authorization failed"; then
+      echo "Test failed: deleting $TOPIC should have been denied"
+      exit 1
+    fi
+
+    set -e
+
+    sent_message="Hello Stackable!"
+
+    # kcat (librdkafka) cannot query offsets on Kafka 4.x, so use the Kafka tools to produce and
+    # consume.
+    echo $sent_message | /stackable/kafka/bin/kafka-console-producer.sh \
+      --bootstrap-server $KAFKA \
+      --producer.config $ADMIN_CONFIG \
+      --topic $TOPIC
+
+    echo Sent message: \"$sent_message\"
+
+    # Kafka 3.9 (and newer versions) print a deprecation warning for --consumer.config on stdout, so filter it
+    received_message=$(/stackable/kafka/bin/kafka-console-consumer.sh \
+      --bootstrap-server $KAFKA \
+      --consumer.config $ADMIN_CONFIG \
+      --group $CONSUMER_GROUP \
+      --topic $TOPIC \
+      --from-beginning \
+      --max-messages 1 \
+      --timeout-ms 30000 | grep -v "is deprecated")
+
+    echo Received message: \"$received_message\"
+
+    if [ "$received_message" = "$sent_message" ]; then
+      echo "Test passed"
+    else
+      echo "Test failed"
+      exit 1
+    fi
+
+    # Temporarily disable exit-on-error
+    set +e
+
+    # This should fail as the rego rule filters topic access
+    output=$(echo "$sent_message" | /stackable/kafka/bin/kafka-console-producer.sh \
+      --bootstrap-server $KAFKA \
+      --producer.config $ADMIN_CONFIG \
+      --topic $DISALLOWED_TOPIC 2>&1)
+    exit_code=$?
+
+    set -e
+
+    if [ $exit_code -ne 0 ] || echo "$output" | grep -qi "authorization failed"; then
+      echo "Test passed"
+      exit 0
+    else
+      echo "Test failed"
+      exit 1
+    fi
diff --git a/tests/templates/kuttl/opa-kraft/40-access-kafka.yaml b/tests/templates/kuttl/opa-kraft/40-access-kafka.yaml
new file mode 100644
index 000000000..966e1ebce
--- /dev/null
+++ b/tests/templates/kuttl/opa-kraft/40-access-kafka.yaml
@@ -0,0 +1,6 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestStep
+commands:
+  # We need to replace $NAMESPACE (by KUTTL)
+  - script: envsubst '$NAMESPACE' < 40-access-kafka.txt | kubectl apply -n $NAMESPACE -f -
diff --git a/tests/templates/kuttl/opa-kraft/40-assert.yaml b/tests/templates/kuttl/opa-kraft/40-assert.yaml
new file mode 100644
index 000000000..edc6c317f
--- /dev/null
+++ b/tests/templates/kuttl/opa-kraft/40-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/opa-kraft/vector-aggregator-values.yaml.j2 b/tests/templates/kuttl/opa-kraft/vector-aggregator-values.yaml.j2
new file mode 100644
index 000000000..1eacd03f2
--- /dev/null
+++ b/tests/templates/kuttl/opa-kraft/vector-aggregator-values.yaml.j2
@@ -0,0 +1,82 @@
+---
+role: Aggregator
+service:
+  ports:
+  - name: api
+    port: 8686
+    protocol: TCP
+    targetPort: 8686
+  - name: vector
+    port: 6123
+    protocol: TCP
+    targetPort: 6000
+customConfig:
+  api:
+    address: 0.0.0.0:8686
+    enabled: true
+  sources:
+    vector:
+      address: 0.0.0.0:6000
+      type: vector
+      version: "2"
+  transforms:
+    validEvents:
+      type: filter
+      inputs: [vector]
+      condition: is_null(.errors)
+    filteredAutomaticLogConfigServerOpaDecision:
+      type: filter
+      inputs: [validEvents]
+      condition: >-
+        starts_with(string!(.pod), "test-opa-server-automatic-log-config") &&
+        .container == "opa" &&
+        .logger == "decision"
+    filteredAutomaticLogConfigServerOpa:
+      type: filter
+      inputs: [validEvents]
+      condition: >-
+        starts_with(string!(.pod), "test-opa-server-automatic-log-config") &&
+        .container == "opa" &&
+        .logger != "decision"
+    filteredAutomaticLogConfigServerBundleBuilder:
+      type: filter
+      inputs: [validEvents]
+      condition: >-
+        starts_with(string!(.pod), "test-opa-server-automatic-log-config") &&
+        .container == "bundle-builder"
+    filteredAutomaticLogConfigServerVector:
+      type: filter
+      inputs: [validEvents]
+      condition: >-
+        starts_with(string!(.pod), "test-opa-server-automatic-log-config") &&
+        .container == "vector"
+    filteredAutomaticLogConfigServerPrepare:
+      type: filter
+      inputs: [validEvents]
+      condition: >-
+        starts_with(string!(.pod), "test-opa-server-automatic-log-config") &&
+        .container == "prepare"
+    filteredInvalidEvents:
+      type: filter
+      inputs: [vector]
+      condition: |-
+        .timestamp == from_unix_timestamp!(0) ||
+        is_null(.level) ||
+        is_null(.logger) ||
+        is_null(.message)
+  sinks:
+    test:
+      inputs: [filtered*]
+      type: blackhole
+{% if lookup('env', 'VECTOR_AGGREGATOR') %}
+    aggregator:
+      inputs: [vector]
+      type: vector
+      routing:
+        endpoints:
+          - {{ lookup('env', 'VECTOR_AGGREGATOR') }}
+      buffer:
+        # Avoid back pressure from VECTOR_AGGREGATOR. The test should
+        # not fail if the aggregator is not available.
+        when_full: drop_newest
+{% endif %}
diff --git a/tests/test-definition.yaml b/tests/test-definition.yaml
index c4c294ffd..38dd572cd 100644
--- a/tests/test-definition.yaml
+++ b/tests/test-definition.yaml
@@ -95,6 +95,15 @@ tests:
       - use-opa-tls
       - opa-latest
       - krb5
+  # Tests OPA authorization with KRaft controllers, which have to authorize the requests
+  # that the brokers forward to them (e.g. creating and deleting topics).
+  - name: opa-kraft
+    dimensions:
+      - kafka-kraft
+      - openshift
+      - use-opa-tls
+      - opa-latest
+      - krb5
   - name: configuration
     dimensions:
       - kafka-latest

From 3d09022da0070bb7c5c5bcec8d0970480c605471 Mon Sep 17 00:00:00 2001
From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com>
Date: Mon, 2 Nov 2026 11:02:54 +0100
Subject: [PATCH 02/13] kuttl: nightly suite enables tls everywhere

---
 tests/test-definition.yaml | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/tests/test-definition.yaml b/tests/test-definition.yaml
index 38dd572cd..7efe4f22d 100644
--- a/tests/test-definition.yaml
+++ b/tests/test-definition.yaml
@@ -38,16 +38,16 @@ dimensions:
       - 4.3.1
   - name: use-client-tls
     values:
-      - "true"
       - "false"
+      - "true"
   - name: use-client-auth-tls
     values:
-      - "true"
       - "false"
+      - "true"
   - name: use-opa-tls
     values:
-      - "true"
       - "false"
+      - "true"
   - name: openshift
     values:
       - "false"

From 02ebf127f9ecedb6d09ab606d0c9e322165a6a96 Mon Sep 17 00:00:00 2001
From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com>
Date: Mon, 2 Nov 2026 12:17:03 +0100
Subject: [PATCH 03/13] tests(configuration): wait for Kafka cluster to
 actually become ready

---
 tests/templates/kuttl/configuration/10-assert.yaml.j2        | 2 ++
 tests/templates/kuttl/configuration/10-install-kafka.yaml.j2 | 1 +
 2 files changed, 3 insertions(+)

diff --git a/tests/templates/kuttl/configuration/10-assert.yaml.j2 b/tests/templates/kuttl/configuration/10-assert.yaml.j2
index 3d75ac23f..507b9c6d2 100644
--- a/tests/templates/kuttl/configuration/10-assert.yaml.j2
+++ b/tests/templates/kuttl/configuration/10-assert.yaml.j2
@@ -2,6 +2,8 @@
 apiVersion: kuttl.dev/v1beta1
 kind: TestAssert
 timeout: 300
+commands:
+  - script: kubectl -n $NAMESPACE wait --for=condition=available=true kafkaclusters.kafka.stackable.tech/test-kafka --timeout 601s
 ---
 apiVersion: apps/v1
 kind: StatefulSet
diff --git a/tests/templates/kuttl/configuration/10-install-kafka.yaml.j2 b/tests/templates/kuttl/configuration/10-install-kafka.yaml.j2
index af9f97e80..7edb23b3e 100644
--- a/tests/templates/kuttl/configuration/10-install-kafka.yaml.j2
+++ b/tests/templates/kuttl/configuration/10-install-kafka.yaml.j2
@@ -17,6 +17,7 @@ spec:
 {% endif %}
     pullPolicy: IfNotPresent
   clusterConfig:
+    metadataManager: KRaft
 {% if lookup('env', 'VECTOR_AGGREGATOR') %}
   clusterConfig:
     vectorAggregatorConfigMapName: vector-aggregator-discovery

From e0df1f422621f7a412df3737ce425bdb235353a8 Mon Sep 17 00:00:00 2001
From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com>
Date: Mon, 2 Nov 2026 16:08:20 +0100
Subject: [PATCH 04/13] docs: describe how internal authorization works

---
 .../security/02-authorization-opa.yaml        | 63 -------------------
 .../kafka/pages/usage-guide/security.adoc     | 28 ++++++---
 .../src/controller/build/properties/mod.rs    | 19 ++++++
 3 files changed, 39 insertions(+), 71 deletions(-)
 delete mode 100644 docs/modules/kafka/examples/security/02-authorization-opa.yaml

diff --git a/docs/modules/kafka/examples/security/02-authorization-opa.yaml b/docs/modules/kafka/examples/security/02-authorization-opa.yaml
deleted file mode 100644
index 754bf5525..000000000
--- a/docs/modules/kafka/examples/security/02-authorization-opa.yaml
+++ /dev/null
@@ -1,63 +0,0 @@
----
-apiVersion: opa.stackable.tech/v1alpha1
-kind: OpaCluster
-metadata:
-  name: simple-opa
-spec:
-  image:
-    productVersion: 1.20.2
-  servers:
-    config:
-      logging:
-        containers:
-          opa:
-            loggers:
-              decision:
-                level: INFO
-    roleGroups:
-      default: {}
----
-apiVersion: v1
-kind: ConfigMap
-metadata:
-  name: kafka-rego-rules
-  labels:
-    opa.stackable.tech/bundle: "true"
-data:
-  kafka.rego: |
-    package kafka
-
-    is_internal_request if input.requestContext.listenerName == "INTERNAL"
-    is_external_request if not is_internal_request
-
-    # FQDNs of the brokers as regular expressions
-    brokers_fqdn := [
-      `local`,                                 # cluster domain
-      `cluster`,                               # cluster domain
-      `svc`,
-      `default`,                               # namespace
-      `simple-kafka-broker-default-headless`,  # Pod subdomain
-      `simple-kafka-broker-default-[0-9]+`,    # Pod name
-    ]
-
-    domain_components = [concat("", [`DC=`, dc]) | some dc in brokers_fqdn]
-    common_name = `CN=generated certificate for pod`
-    subject_dn_fields = array.concat(domain_components, [common_name])
-    brokers_subject_dn_pattern := concat("", ["^", concat(",", subject_dn_fields), "$"])
-
-    default allow := false
-
-    allow if {
-      is_internal_request
-
-      regex.match(
-        brokers_subject_dn_pattern,
-        input.requestContext.principal.name,
-      )
-    }
-
-    allow if {
-      is_external_request
-
-      # TODO Add your own rules
-    }
diff --git a/docs/modules/kafka/pages/usage-guide/security.adoc b/docs/modules/kafka/pages/usage-guide/security.adoc
index 977803cae..347994c8a 100644
--- a/docs/modules/kafka/pages/usage-guide/security.adoc
+++ b/docs/modules/kafka/pages/usage-guide/security.adoc
@@ -231,12 +231,24 @@ include::example$security/03b-authorization-kafkacluster-custom-properties.yaml[
 
 A full list of settings and their respective defaults can be found https://github.com/anderseknert/opa-kafka-plugin[here].
 
-The OPA Rego rules can authorize both external and internal communication.
-For internal communication, the principal name is the subject DN of the broker's certificate.
-The subject DN contains the FQDN of the pod and a generic common name, e.g. `DC=local,DC=cluster,DC=svc,DC=default,DC=simple-kafka-broker-default-headless,DC=simple-kafka-broker-default-0,CN=generated certificate for pod`.
-An OpaCluster and the Rego rules that verify the subject DNs could look as follows:
+The OPA Rego rules authorize requests from external clients.
+Communication between the Kafka nodes is not authorized by OPA, see <<authorization-internal-traffic>>.
 
-[source,yaml]
-----
-include::example$security/02-authorization-opa.yaml[]
-----
+=== [[authorization-internal-traffic]]Internal traffic
+
+The Kafka nodes (brokers and controllers) are configured as `super.users`, so that the traffic between them is allowed without asking OPA.
+You do not need to write Rego rules for it.
+Requests that a broker forwards to a controller on behalf of a client, such as creating a topic, are still authorized against the principal of the original client.
+
+The operator sets `super.users` to the principals of all nodes of the cluster:
+
+* The nodes authenticate each other with the TLS certificates that the secret operator provides for the `internalSecretClass`.
+  The principal is the subject DN of the certificate, which contains the FQDN of the pod and a generic common name, e.g. `DC=local,DC=cluster,DC=svc,DC=default,DC=simple-kafka-broker-default-headless,DC=simple-kafka-broker-default-0,CN=generated certificate for pod`.
+* If Kerberos is enabled, the short names of the Kerberos services of the brokers and the controllers are added as well.
+
+Kafka compares the principal of a request with each `super.users` entry for exact equality; wildcards are not supported.
+The common name `CN=generated certificate for pod` is therefore part of every certificate-based entry.
+The secret operator always sets it in the subject of the pod certificates.
+
+NOTE: Everything that presents a certificate with the subject DN of a node, issued by the CA of the `internalSecretClass`, is treated as a Kafka node.
+Restrict who can obtain such certificates accordingly.
diff --git a/rust/operator-binary/src/controller/build/properties/mod.rs b/rust/operator-binary/src/controller/build/properties/mod.rs
index e3de0569a..582fb9bd3 100644
--- a/rust/operator-binary/src/controller/build/properties/mod.rs
+++ b/rust/operator-binary/src/controller/build/properties/mod.rs
@@ -153,6 +153,25 @@ pub(crate) fn opa_authorizer_properties(
 /// (mTLS), whose subject DN is `CN=generated certificate for pod,DC=<pod>,DC=<service>,
 /// DC=<namespace>,DC=svc,DC=<cluster domain parts>`, see the secret-operator.
 /// With Kerberos, the node principals are mapped to the short name of the service.
+///
+/// # Why the `CN=generated certificate for pod` entry is required
+///
+/// The secret-operator always adds this common name to the subject of the pod certificates,
+/// regardless of whether domain components are added as well. Kafka's default principal for an
+/// mTLS client is the complete subject DN in RFC 2253 notation, which lists the RDNs in reverse
+/// order, so the CN is the last component: `DC=local,...,DC=<pod>,CN=generated certificate for pod`.
+///
+/// Kafka compares each `super.users` entry for exact equality with the principal; there is no
+/// support for wildcards or prefixes (the only wildcard Kafka knows is `User:*` in ACLs, which
+/// matches all users). An entry without the CN, or with a `*` in place of it, therefore never
+/// matches, and the brokers and controllers would be denied access to each other.
+///
+/// This could only be avoided by rewriting the principal with `ssl.principal.mapping.rules`,
+/// which is not done here.
+///
+/// See the Kafka documentation:
+/// - [Authorization and ACLs](https://kafka.apache.org/44/security/authorization-and-acls/)
+/// - [`ssl.principal.mapping.rules`](https://kafka.apache.org/44/security/authorization-and-acls/#customizing-ssl-user-name)
 fn super_users(pod_descriptors: &[KafkaPodDescriptor], kerberos_enabled: bool) -> String {
     let mut principals: BTreeSet<String> = pod_descriptors
         .iter()

From f0c0db33ed303d42e655a39b6cc9d5e101785e2f Mon Sep 17 00:00:00 2001
From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com>
Date: Mon, 2 Nov 2026 16:44:10 +0100
Subject: [PATCH 05/13] Add OPA pod affinity to controller pods.

---
 CHANGELOG.md                                  |  3 +-
 .../usage-guide/operations/pod-placement.adoc |  6 +-
 .../src/controller/validate.rs                |  1 +
 rust/operator-binary/src/crd/affinity.rs      | 62 +++++++++----------
 .../src/crd/role/controller.rs                | 19 +++---
 5 files changed, 46 insertions(+), 45 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index fc2d0efd0..0492b5a38 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -15,7 +15,8 @@ All notable changes to this project will be documented in this file.
 - Brokers now have a default affinity to the OPA Pods when OPA authorization is configured ([#1035]).
 - The OPA authorizer is now also configured on KRaft controllers. Previously, requests that brokers
   forward to the controllers (e.g. creating or deleting topics) were not authorized at all.
-  The Kafka nodes are super users, so that traffic between them does not hit OPA ([#1051]).
+  The Kafka nodes are now configured as super users, so that traffic between them does not hit OPA.
+  Controllers now get the same affinity to OPA as brokers ([#1051]).
 
 ### Changed
 
diff --git a/docs/modules/kafka/pages/usage-guide/operations/pod-placement.adoc b/docs/modules/kafka/pages/usage-guide/operations/pod-placement.adoc
index 0a03be18c..2035fb271 100644
--- a/docs/modules/kafka/pages/usage-guide/operations/pod-placement.adoc
+++ b/docs/modules/kafka/pages/usage-guide/operations/pod-placement.adoc
@@ -1,6 +1,6 @@
 = Pod placement
 
-You can configure Pod placement for Kafka brokers as described in xref:concepts:pod_placement.adoc[].
+You can configure Pod placement for Kafka brokers and controllers as described in xref:concepts:pod_placement.adoc[].
 
 By default, the operator configures the following Pod placement constraints:
 
@@ -21,8 +21,8 @@ affinity:
 
 In the example above `cluster-name` is the name of the Kafka custom resource that owns this Pod.
 
-If OPA authorization is configured, the broker Pods additionally prefer to run on the same Kubernetes nodes as the OPA Pods (weight 50).
-Only brokers are configured with the OPA authorizer, as they are the only role issuing OPA requests.
+If OPA authorization is configured, the broker and controller Pods additionally prefer to run on the same Kubernetes nodes as the OPA Pods (weight 50).
+Both roles run the OPA authorizer, so both issue OPA requests.
 
 [source,yaml]
 ----
diff --git a/rust/operator-binary/src/controller/validate.rs b/rust/operator-binary/src/controller/validate.rs
index 3f0f2322d..634a21b52 100644
--- a/rust/operator-binary/src/controller/validate.rs
+++ b/rust/operator-binary/src/controller/validate.rs
@@ -300,6 +300,7 @@ pub fn validate(
                 ControllerConfig::default_config(
                     &kafka.name_any(),
                     &KafkaRole::Controller.to_string(),
+                    kafka.spec.cluster_config.authorization.opa.as_ref(),
                 ),
                 cluster_id,
                 &vector_aggregator_config_map_name,
diff --git a/rust/operator-binary/src/crd/affinity.rs b/rust/operator-binary/src/crd/affinity.rs
index e46b71ecc..dda570a16 100644
--- a/rust/operator-binary/src/crd/affinity.rs
+++ b/rust/operator-binary/src/crd/affinity.rs
@@ -8,7 +8,8 @@ use stackable_operator::{
 
 use crate::crd::APP_NAME;
 
-/// `opa_config` is only passed for roles that send authorization requests to OPA.
+/// `opa_config` is only passed if OPA authorization is configured. Brokers and controllers
+/// both run the OPA authorizer, so both send authorization requests to OPA.
 pub fn get_affinity(
     cluster_name: &str,
     role: &str,
@@ -112,39 +113,32 @@ mod tests {
         assert_eq!(
             *affinity,
             StackableAffinity {
-                pod_affinity: match role {
-                    // Only brokers are configured with the OPA authorizer.
-                    KafkaRole::Broker => Some(PodAffinity {
-                        preferred_during_scheduling_ignored_during_execution: Some(vec![
-                            WeightedPodAffinityTerm {
-                                pod_affinity_term: PodAffinityTerm {
-                                    label_selector: Some(LabelSelector {
-                                        match_expressions: None,
-                                        match_labels: Some(BTreeMap::from([
-                                            (
-                                                "app.kubernetes.io/name".to_string(),
-                                                "opa".to_string()
-                                            ),
-                                            (
-                                                "app.kubernetes.io/instance".to_string(),
-                                                "simple-opa".to_string(),
-                                            ),
-                                            (
-                                                "app.kubernetes.io/component".to_string(),
-                                                "server".to_string(),
-                                            ),
-                                        ])),
-                                    }),
-                                    topology_key: "kubernetes.io/hostname".to_string(),
-                                    ..PodAffinityTerm::default()
-                                },
-                                weight: 50,
-                            }
-                        ]),
-                        required_during_scheduling_ignored_during_execution: None,
-                    }),
-                    KafkaRole::Controller => None,
-                },
+                pod_affinity: Some(PodAffinity {
+                    preferred_during_scheduling_ignored_during_execution: Some(vec![
+                        WeightedPodAffinityTerm {
+                            pod_affinity_term: PodAffinityTerm {
+                                label_selector: Some(LabelSelector {
+                                    match_expressions: None,
+                                    match_labels: Some(BTreeMap::from([
+                                        ("app.kubernetes.io/name".to_string(), "opa".to_string()),
+                                        (
+                                            "app.kubernetes.io/instance".to_string(),
+                                            "simple-opa".to_string(),
+                                        ),
+                                        (
+                                            "app.kubernetes.io/component".to_string(),
+                                            "server".to_string(),
+                                        ),
+                                    ])),
+                                }),
+                                topology_key: "kubernetes.io/hostname".to_string(),
+                                ..PodAffinityTerm::default()
+                            },
+                            weight: 50,
+                        }
+                    ]),
+                    required_during_scheduling_ignored_during_execution: None,
+                }),
                 pod_anti_affinity: Some(PodAntiAffinity {
                     preferred_during_scheduling_ignored_during_execution: Some(vec![
                         WeightedPodAffinityTerm {
diff --git a/rust/operator-binary/src/crd/role/controller.rs b/rust/operator-binary/src/crd/role/controller.rs
index 58540482b..e13a0114d 100644
--- a/rust/operator-binary/src/crd/role/controller.rs
+++ b/rust/operator-binary/src/crd/role/controller.rs
@@ -2,9 +2,12 @@ use std::str::FromStr;
 
 use serde::{Deserialize, Serialize};
 use stackable_operator::{
-    commons::resources::{
-        CpuLimitsFragment, MemoryLimitsFragment, NoRuntimeLimits, NoRuntimeLimitsFragment,
-        PvcConfigFragment, Resources, ResourcesFragment,
+    commons::{
+        opa::OpaConfig,
+        resources::{
+            CpuLimitsFragment, MemoryLimitsFragment, NoRuntimeLimits, NoRuntimeLimitsFragment,
+            PvcConfigFragment, Resources, ResourcesFragment,
+        },
     },
     config::{fragment::Fragment, merge::Merge},
     constant,
@@ -78,11 +81,13 @@ pub struct ControllerConfig {
 }
 
 impl ControllerConfig {
-    pub fn default_config(cluster_name: &str, role: &str) -> ControllerConfigFragment {
+    pub fn default_config(
+        cluster_name: &str,
+        role: &str,
+        opa_config: Option<&OpaConfig>,
+    ) -> ControllerConfigFragment {
         ControllerConfigFragment {
-            // Only brokers are configured with the OPA authorizer, so controllers get no affinity
-            // to the OPA Pods.
-            common_config: CommonConfig::default_config(cluster_name, role, None),
+            common_config: CommonConfig::default_config(cluster_name, role, opa_config),
             logging: product_logging::spec::default_logging(),
             resources: ResourcesFragment {
                 cpu: CpuLimitsFragment {

From b177e24a9731c8382549d42cba39823006ba0f7b Mon Sep 17 00:00:00 2001
From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com>
Date: Tue, 3 Nov 2026 16:09:25 +0100
Subject: [PATCH 06/13] review: rename function

---
 rust/operator-binary/src/controller/build/security.rs | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/rust/operator-binary/src/controller/build/security.rs b/rust/operator-binary/src/controller/build/security.rs
index 4e69e763d..a6ce91a26 100644
--- a/rust/operator-binary/src/controller/build/security.rs
+++ b/rust/operator-binary/src/controller/build/security.rs
@@ -260,7 +260,7 @@ pub fn controller_admin_client_properties(
 /// the pod volume it mounts.
 ///
 /// The CA is imported into the internal truststore on startup, see [`copy_opa_tls_cert_command`].
-fn add_opa_tls_volume_mount(
+fn mount_opa_tls_volume(
     security: &ValidatedKafkaSecurity,
     cb_kafka: &mut ContainerBuilder,
 ) -> Result<Option<Volume>, Error> {
@@ -339,7 +339,7 @@ pub fn mount_broker_tls_volumes(
         )
         .expect("The mount paths are statically defined and there should be no duplicates.");
 
-    volumes.extend(add_opa_tls_volume_mount(security, cb_kafka)?);
+    volumes.extend(mount_opa_tls_volume(security, cb_kafka)?);
 
     Ok(volumes)
 }
@@ -380,7 +380,7 @@ pub fn mount_controller_tls_volume(
         .expect("The mount paths are statically defined and there should be no duplicates.");
 
     let mut volumes = vec![volume];
-    volumes.extend(add_opa_tls_volume_mount(security, cb_kafka)?);
+    volumes.extend(mount_opa_tls_volume(security, cb_kafka)?);
 
     Ok(volumes)
 }

From d08702479295e1c6dcdef5e9c1d19635301cd7a9 Mon Sep 17 00:00:00 2001
From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com>
Date: Tue, 3 Nov 2026 16:16:35 +0100
Subject: [PATCH 07/13] review: replace extends() call with explicit push()

---
 rust/operator-binary/src/controller/build/security.rs | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/rust/operator-binary/src/controller/build/security.rs b/rust/operator-binary/src/controller/build/security.rs
index a6ce91a26..5568b1257 100644
--- a/rust/operator-binary/src/controller/build/security.rs
+++ b/rust/operator-binary/src/controller/build/security.rs
@@ -339,7 +339,9 @@ pub fn mount_broker_tls_volumes(
         )
         .expect("The mount paths are statically defined and there should be no duplicates.");
 
-    volumes.extend(mount_opa_tls_volume(security, cb_kafka)?);
+    if let Some(opa_tls_volume) = mount_opa_tls_volume(security, cb_kafka)? {
+        volumes.push(opa_tls_volume);
+    }
 
     Ok(volumes)
 }

From afdfb177a47d7b138276eafd2042fe12494c7a48 Mon Sep 17 00:00:00 2001
From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com>
Date: Tue, 3 Nov 2026 16:29:03 +0100
Subject: [PATCH 08/13] review: split OPA certificate mounting from the
 internal certificate

---
 .../controller/build/role_group/controller.rs   | 17 ++++++++++++++---
 .../src/controller/build/security.rs            | 14 +++++---------
 2 files changed, 19 insertions(+), 12 deletions(-)

diff --git a/rust/operator-binary/src/controller/build/role_group/controller.rs b/rust/operator-binary/src/controller/build/role_group/controller.rs
index a4fb4fc27..af391e36f 100644
--- a/rust/operator-binary/src/controller/build/role_group/controller.rs
+++ b/rust/operator-binary/src/controller/build/role_group/controller.rs
@@ -45,6 +45,7 @@ use crate::{
             security::{
                 self, STACKABLE_TLS_KAFKA_INTERNAL_DIR, STACKABLE_TLS_KAFKA_INTERNAL_VOLUME_NAME,
                 controller_admin_client_properties, mount_controller_tls_volume,
+                mount_opa_tls_volume,
             },
         },
         node_id_hasher::node_id_hash32_offset,
@@ -80,7 +81,7 @@ pub enum Error {
     BuildPodDescriptors { source: PodDescriptorsError },
 
     #[snafu(display("failed to build the TLS volume"))]
-    BuildTlsVolumes { source: security::Error },
+    BuildInternalTlsVolume { source: security::Error },
 
     #[snafu(display("failed to build the Kerberos volume"))]
     BuildKerberosVolume { source: kerberos::Error },
@@ -90,6 +91,9 @@ pub enum Error {
 
     #[snafu(display("failed to construct JVM arguments"))]
     ConstructJvmArguments { source: jvm::Error },
+
+    #[snafu(display("failed to build the OPA TLS volume"))]
+    BuildOpaTlsVolume { source: security::Error },
 }
 
 // Env vars the operator sets on the controller's `kafka` and `quorum-manager` containers.
@@ -332,15 +336,22 @@ fn role_pod(
     pod_volumes.push(log_config_volume(logging, &resource_names));
 
     // The TLS certificate is requested with the role group's secret lifetime.
-    pod_volumes.extend(
+    pod_volumes.push(
         mount_controller_tls_volume(
             kafka_security,
             &mut cb_kafka,
             &config.requested_secret_lifetime,
         )
-        .context(BuildTlsVolumesSnafu)?,
+        .context(BuildInternalTlsVolumeSnafu)?,
     );
 
+    // Mount the OPA TLS certificate if any.
+    if let Some(opa_tls_volume) =
+        mount_opa_tls_volume(kafka_security, &mut cb_kafka).context(BuildOpaTlsVolumeSnafu)?
+    {
+        pod_volumes.push(opa_tls_volume);
+    }
+
     let containers = vec![
         cb_kafka.build(),
         build_quorum_manager_container(image, kafka_security, quorum_manager_env),
diff --git a/rust/operator-binary/src/controller/build/security.rs b/rust/operator-binary/src/controller/build/security.rs
index 5568b1257..ba1d1a8e2 100644
--- a/rust/operator-binary/src/controller/build/security.rs
+++ b/rust/operator-binary/src/controller/build/security.rs
@@ -260,7 +260,7 @@ pub fn controller_admin_client_properties(
 /// the pod volume it mounts.
 ///
 /// The CA is imported into the internal truststore on startup, see [`copy_opa_tls_cert_command`].
-fn mount_opa_tls_volume(
+pub fn mount_opa_tls_volume(
     security: &ValidatedKafkaSecurity,
     cb_kafka: &mut ContainerBuilder,
 ) -> Result<Option<Volume>, Error> {
@@ -346,18 +346,17 @@ pub fn mount_broker_tls_volumes(
     Ok(volumes)
 }
 
-/// Adds the internal TLS volume mount to the controller's container builder and returns the pod
-/// volumes they mount.
+/// Adds the internal TLS volume mount to the controller's container builder.
 ///
 /// # Panics
 ///
-/// Panics if the volume mounts cannot be added to the container builder. Only call this on a
+/// Panics if the volume mount cannot be added to the container builder. Only call this on a
 /// container builder whose mount paths are still distinct from the ones added here.
 pub fn mount_controller_tls_volume(
     security: &ValidatedKafkaSecurity,
     cb_kafka: &mut ContainerBuilder,
     requested_secret_lifetime: &Duration,
-) -> Result<Vec<Volume>, Error> {
+) -> Result<Volume, Error> {
     let volume = VolumeBuilder::new(&*STACKABLE_TLS_KAFKA_INTERNAL_VOLUME_NAME)
         .ephemeral(
             SecretOperatorVolumeSourceBuilder::new(
@@ -381,10 +380,7 @@ pub fn mount_controller_tls_volume(
         )
         .expect("The mount paths are statically defined and there should be no duplicates.");
 
-    let mut volumes = vec![volume];
-    volumes.extend(mount_opa_tls_volume(security, cb_kafka)?);
-
-    Ok(volumes)
+    Ok(volume)
 }
 
 /// Inserts the `listener.<name>.ssl.{keystore,truststore}.{location,password,type}`

From 0b97a05ff2816bbb8b3a567405fc583c901df0ae Mon Sep 17 00:00:00 2001
From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com>
Date: Tue, 3 Nov 2026 16:54:22 +0100
Subject: [PATCH 09/13] review: make sure that
 listener.name.controller.ssl.client.auth is set only once

---
 .../build/properties/controller_properties.rs | 18 +++-------
 .../src/controller/build/security.rs          | 36 ++++++++++++++-----
 2 files changed, 33 insertions(+), 21 deletions(-)

diff --git a/rust/operator-binary/src/controller/build/properties/controller_properties.rs b/rust/operator-binary/src/controller/build/properties/controller_properties.rs
index 128a7ef91..c5de74c62 100644
--- a/rust/operator-binary/src/controller/build/properties/controller_properties.rs
+++ b/rust/operator-binary/src/controller/build/properties/controller_properties.rs
@@ -6,7 +6,7 @@ use crate::{
         ValidatedClusterConfig,
         build::{
             graceful_shutdown::graceful_shutdown_config_properties,
-            security::{SSL_CLIENT_AUTH_REQUIRED, controller_config_settings},
+            security::controller_config_settings,
         },
     },
     crd::{
@@ -74,18 +74,10 @@ pub fn build(
     }
 
     result.extend(opa_authorizer_properties(cluster_config, pod_descriptors));
-    if cluster_config.opa_connect().is_some()
-        && !cluster_config.kafka_security.has_kerberos_enabled()
-    {
-        // The nodes are identified as super users by the subject of their internal certificate.
-        // Without client authentication, the principal on the CONTROLLER listener is ANONYMOUS.
-        // With Kerberos, the principal is derived from the SASL authentication instead.
-        result.insert(
-            KafkaListenerName::Controller.listener_ssl_client_auth(),
-            SSL_CLIENT_AUTH_REQUIRED.to_string(),
-        );
-    }
-    result.extend(controller_config_settings(&cluster_config.kafka_security));
+    result.extend(controller_config_settings(
+        &cluster_config.kafka_security,
+        cluster_config.opa_connect().is_some(),
+    ));
     result.extend(graceful_shutdown_config_properties());
     result.extend(overrides);
 
diff --git a/rust/operator-binary/src/controller/build/security.rs b/rust/operator-binary/src/controller/build/security.rs
index ba1d1a8e2..fce6fceae 100644
--- a/rust/operator-binary/src/controller/build/security.rs
+++ b/rust/operator-binary/src/controller/build/security.rs
@@ -560,7 +560,10 @@ pub fn broker_config_settings(security: &ValidatedKafkaSecurity) -> BTreeMap<Str
 
 /// Returns required Kafka configuration settings for the `controller.properties` file
 /// depending on the tls and authentication settings.
-pub fn controller_config_settings(security: &ValidatedKafkaSecurity) -> BTreeMap<String, String> {
+pub fn controller_config_settings(
+    security: &ValidatedKafkaSecurity,
+    opa_authorization: bool,
+) -> BTreeMap<String, String> {
     let mut config = BTreeMap::new();
 
     insert_listener_ssl_stores(
@@ -579,7 +582,13 @@ pub fn controller_config_settings(security: &ValidatedKafkaSecurity) -> BTreeMap
     // We set either client tls with authentication or client tls without authentication
     // If authentication is explicitly required we do not want to have any other CAs to
     // be trusted.
-    if security.tls_client_authentication_class().is_some() {
+    //
+    // With OPA, the nodes are identified as super users by the subject of their internal
+    // certificate. Without client authentication, the principal on the CONTROLLER listener is
+    // ANONYMOUS. With Kerberos, the principal is derived from the SASL authentication instead.
+    if security.tls_client_authentication_class().is_some()
+        || (opa_authorization && !security.has_kerberos_enabled())
+    {
         // client auth required
         config.insert(
             KafkaListenerName::Controller.listener_ssl_client_auth(),
@@ -1167,7 +1176,7 @@ pub(crate) mod tests {
 
     #[test]
     fn controller_config_opa_adds_authorizer_truststore() {
-        let config = controller_config_settings(&opa());
+        let config = controller_config_settings(&opa(), false);
         assert!(config.contains_key("opa.authorizer.truststore.path"));
         assert!(config.contains_key("opa.authorizer.truststore.password"));
         assert!(config.contains_key("opa.authorizer.truststore.type"));
@@ -1178,7 +1187,7 @@ pub(crate) mod tests {
         // Internal TLS is mandatory, so the controller config is non-empty even for a plaintext
         // cluster: it carries the controller + internal SSL stores, but no client-cert auth
         // (there is no client authentication class).
-        let config = controller_config_settings(&plaintext());
+        let config = controller_config_settings(&plaintext(), false);
         assert!(
             config.contains_key(&KafkaListenerName::Controller.listener_ssl_keystore_location())
         );
@@ -1186,9 +1195,20 @@ pub(crate) mod tests {
         assert!(!config.contains_key(&KafkaListenerName::Controller.listener_ssl_client_auth()));
     }
 
+    #[test]
+    fn controller_config_opa_requires_client_auth_unless_kerberos() {
+        let key = KafkaListenerName::Controller.listener_ssl_client_auth();
+
+        let config = controller_config_settings(&internal_tls(), true);
+        assert_eq!(config.get(&key).map(String::as_str), Some("required"));
+
+        assert!(!controller_config_settings(&internal_tls(), false).contains_key(&key));
+        assert!(!controller_config_settings(&kerberos(), true).contains_key(&key));
+    }
+
     #[test]
     fn controller_config_internal_tls_covers_controller_and_internal() {
-        let config = controller_config_settings(&internal_tls());
+        let config = controller_config_settings(&internal_tls(), false);
         assert!(config.contains_key("listener.name.controller.ssl.keystore.location"));
         assert!(config.contains_key("listener.name.internal.ssl.keystore.location"));
     }
@@ -1204,7 +1224,7 @@ pub(crate) mod tests {
 
     #[test]
     fn controller_config_sets_the_controller_sasl_mechanism_with_kerberos() {
-        let config = controller_config_settings(&kerberos());
+        let config = controller_config_settings(&kerberos(), false);
         assert_eq!(
             config.get("sasl.mechanism.controller.protocol"),
             Some(&"GSSAPI".to_string())
@@ -1218,14 +1238,14 @@ pub(crate) mod tests {
                 .contains_key("sasl.mechanism.controller.protocol")
         );
         assert!(
-            !controller_config_settings(&internal_tls())
+            !controller_config_settings(&internal_tls(), false)
                 .contains_key("sasl.mechanism.controller.protocol")
         );
     }
 
     #[test]
     fn controller_config_kerberos_adds_sasl() {
-        let config = controller_config_settings(&kerberos());
+        let config = controller_config_settings(&kerberos(), false);
         assert_eq!(
             config.get("sasl.enabled.mechanisms"),
             Some(&"GSSAPI".to_string())

From 3c05f94d0a8703c263912c4b13c59e3dfbc7374d Mon Sep 17 00:00:00 2001
From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com>
Date: Tue, 3 Nov 2026 16:58:20 +0100
Subject: [PATCH 10/13] review: remove duplicate yaml key in test

---
 tests/templates/kuttl/configuration/10-install-kafka.yaml.j2 | 1 -
 1 file changed, 1 deletion(-)

diff --git a/tests/templates/kuttl/configuration/10-install-kafka.yaml.j2 b/tests/templates/kuttl/configuration/10-install-kafka.yaml.j2
index 7edb23b3e..c026bdaae 100644
--- a/tests/templates/kuttl/configuration/10-install-kafka.yaml.j2
+++ b/tests/templates/kuttl/configuration/10-install-kafka.yaml.j2
@@ -19,7 +19,6 @@ spec:
   clusterConfig:
     metadataManager: KRaft
 {% if lookup('env', 'VECTOR_AGGREGATOR') %}
-  clusterConfig:
     vectorAggregatorConfigMapName: vector-aggregator-discovery
 {% endif %}
   controllers:

From 816cd02314c4c85686324dd7cdcfeba977aaed8f Mon Sep 17 00:00:00 2001
From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com>
Date: Tue, 3 Nov 2026 17:15:53 +0100
Subject: [PATCH 11/13] review: add service_fqdn() and pod_fqdn() for better
 reusability

---
 .../src/controller/build/properties/mod.rs    | 35 ++++---------------
 rust/operator-binary/src/crd/mod.rs           | 26 ++++++++++++++
 2 files changed, 33 insertions(+), 28 deletions(-)

diff --git a/rust/operator-binary/src/controller/build/properties/mod.rs b/rust/operator-binary/src/controller/build/properties/mod.rs
index 582fb9bd3..29dca78a3 100644
--- a/rust/operator-binary/src/controller/build/properties/mod.rs
+++ b/rust/operator-binary/src/controller/build/properties/mod.rs
@@ -83,25 +83,12 @@ pub(crate) fn kraft_controllers(
 
     controllers
         .map(|desc| {
-            if kerberos_enabled {
-                format!(
-                    "{sts}-{replica}.{service}.{namespace}.svc.{cluster_domain}:{client_port}",
-                    sts = desc.role_group_statefulset_name,
-                    replica = desc.replica,
-                    service = desc.role_group_service_name,
-                    namespace = desc.namespace,
-                    cluster_domain = desc.cluster_domain,
-                    client_port = desc.client_port,
-                )
+            let host = if kerberos_enabled {
+                desc.pod_fqdn()
             } else {
-                format!(
-                    "{service}.{namespace}.svc.{cluster_domain}:{client_port}",
-                    service = desc.role_group_service_name,
-                    namespace = desc.namespace,
-                    cluster_domain = desc.cluster_domain,
-                    client_port = desc.client_port,
-                )
-            }
+                desc.service_fqdn()
+            };
+            format!("{host}:{port}", port = desc.client_port)
         })
         .collect::<BTreeSet<_>>()
         .into_iter()
@@ -178,17 +165,9 @@ fn super_users(pod_descriptors: &[KafkaPodDescriptor], kerberos_enabled: bool) -
         .map(|desc| {
             // The certificate encodes `CN`, then the `DC`s from the pod up to the cluster domain
             // suffix. Java prints the subject DN in RFC 2253 order, which is reversed.
-            let cluster_domain = desc.cluster_domain.to_string();
-            let domain_components = [
-                format!("{}-{}", desc.role_group_statefulset_name, desc.replica),
-                desc.role_group_service_name.to_string(),
-                desc.namespace.to_string(),
-                "svc".to_string(),
-            ]
-            .into_iter()
-            .chain(cluster_domain.split('.').map(str::to_string));
+            let pod_fqdn = desc.pod_fqdn();
             let mut rdns: Vec<String> = vec!["CN=generated certificate for pod".to_string()];
-            rdns.extend(domain_components.map(|dc| format!("DC={dc}")));
+            rdns.extend(pod_fqdn.split('.').map(|dc| format!("DC={dc}")));
             rdns.reverse();
             format!("User:{}", rdns.join(","))
         })
diff --git a/rust/operator-binary/src/crd/mod.rs b/rust/operator-binary/src/crd/mod.rs
index 77bbfd17c..388ed5452 100644
--- a/rust/operator-binary/src/crd/mod.rs
+++ b/rust/operator-binary/src/crd/mod.rs
@@ -346,6 +346,32 @@ pub struct KafkaPodDescriptor {
     pub client_port: Port,
 }
 
+impl KafkaPodDescriptor {
+    /// The FQDN of the role group's headless `Service`:
+    /// `<service>.<namespace>.svc.<cluster domain>`.
+    pub(crate) fn service_fqdn(&self) -> String {
+        format!(
+            "{service}.{namespace}.svc.{cluster_domain}",
+            service = self.role_group_service_name,
+            namespace = self.namespace,
+            cluster_domain = self.cluster_domain,
+        )
+    }
+
+    /// The FQDN of the `Pod`: `<statefulset>-<replica>.<service FQDN>`.
+    ///
+    /// The statefulset, service and namespace names are DNS labels and contain no dots, so
+    /// splitting the result on `.` yields the individual labels.
+    pub(crate) fn pod_fqdn(&self) -> String {
+        format!(
+            "{sts}-{replica}.{service_fqdn}",
+            sts = self.role_group_statefulset_name,
+            replica = self.replica,
+            service_fqdn = self.service_fqdn(),
+        )
+    }
+}
+
 #[derive(Clone, Default, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
 #[serde(rename_all = "camelCase")]
 pub struct KafkaClusterStatus {

From 24bd4de6f5c51452f5b95464184b367a3c276845 Mon Sep 17 00:00:00 2001
From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com>
Date: Tue, 3 Nov 2026 17:51:06 +0100
Subject: [PATCH 12/13] review: delete the OPA cert before importing it in the
 truststore

---
 rust/operator-binary/src/controller/build/security.rs | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/rust/operator-binary/src/controller/build/security.rs b/rust/operator-binary/src/controller/build/security.rs
index fce6fceae..dccf549ee 100644
--- a/rust/operator-binary/src/controller/build/security.rs
+++ b/rust/operator-binary/src/controller/build/security.rs
@@ -76,7 +76,11 @@ pub enum Error {
 pub fn copy_opa_tls_cert_command(security: &ValidatedKafkaSecurity) -> String {
     match security.opa_secret_class().is_some() {
         true => format!(
-            "keytool -importcert -file {opa_mount_path}/ca.crt -keystore {tls_dir}/{truststore} -storepass '{tls_password}' -alias opa-ca -noprompt",
+            // The truststore lives on a volume that survives container restarts, so drop any
+            // previously imported alias first. This prevents keytool -import from failing and
+            // leaving the pods in a crash loop.
+            "keytool -delete -alias opa-ca -keystore {tls_dir}/{truststore} -storepass '{tls_password}' 2>/dev/null || true\n\
+             keytool -importcert -file {opa_mount_path}/ca.crt -keystore {tls_dir}/{truststore} -storepass '{tls_password}' -alias opa-ca -noprompt",
             opa_mount_path = OPA_TLS_MOUNT_PATH,
             tls_dir = STACKABLE_TLS_KAFKA_INTERNAL_DIR,
             truststore = TRUSTSTORE_P12_FILE_NAME,

From 7f79c6a7309990d0660545c78dc108e0f5759c41 Mon Sep 17 00:00:00 2001
From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com>
Date: Tue, 3 Nov 2026 19:33:18 +0100
Subject: [PATCH 13/13] review: update docs to clarify scaling behaviour

---
 .../pages/usage-guide/operations/cluster-operations.adoc | 9 +++++++++
 docs/modules/kafka/pages/usage-guide/security.adoc       | 7 +++++++
 2 files changed, 16 insertions(+)

diff --git a/docs/modules/kafka/pages/usage-guide/operations/cluster-operations.adoc b/docs/modules/kafka/pages/usage-guide/operations/cluster-operations.adoc
index bddb9b72c..0a89c476d 100644
--- a/docs/modules/kafka/pages/usage-guide/operations/cluster-operations.adoc
+++ b/docs/modules/kafka/pages/usage-guide/operations/cluster-operations.adoc
@@ -2,3 +2,12 @@
 = Cluster operation
 
 The managed Kafka instances can be configured with different cluster operations like pausing reconciliation or stopping the cluster. See xref:concepts:operations/cluster_operations.adoc[cluster operations] for more details.
+
+== [[scaling-restarts]]Scaling operations restart the whole cluster
+
+Changing the number of replicas of any Kafka role (brokers or controllers) restarts all Pods of the Kafka cluster, not only the Pods of the scaled role.
+
+Scaling a role changes the ConfigMap with the node properties.
+The restart controller detects this change and triggers a restart, which results in all Pods of the cluster being restarted.
+
+Plan scaling operations accordingly, for example by performing them during a maintenance window.
diff --git a/docs/modules/kafka/pages/usage-guide/security.adoc b/docs/modules/kafka/pages/usage-guide/security.adoc
index 347994c8a..b0957bd7d 100644
--- a/docs/modules/kafka/pages/usage-guide/security.adoc
+++ b/docs/modules/kafka/pages/usage-guide/security.adoc
@@ -252,3 +252,10 @@ The secret operator always sets it in the subject of the pod certificates.
 
 NOTE: Everything that presents a certificate with the subject DN of a node, issued by the CA of the `internalSecretClass`, is treated as a Kafka node.
 Restrict who can obtain such certificates accordingly.
+
+If you previously set `super.users` for the Kafka nodes yourself, for example with `configOverrides`, you can remove that override.
+The operator now sets the principals of all nodes automatically.
+Keep an override only if you need additional super users, such as specific client principals, and then include the entries for the nodes as well, because the override replaces the value set by the operator.
+
+NOTE: Scaling a role changes the nodes of the cluster and with them the node properties.
+This restarts all Pods of the cluster, see xref:usage-guide/operations/cluster-operations.adoc#scaling-restarts[Scaling operations restart the whole cluster].