feat: Add affinity to OPA Pods

CHANGELOG.md

@@ -12,6 +12,7 @@ All notable changes to this project will be documented in this file.
   broker-to-controller and controller-to-controller (Raft) traffic ([#1024]).
 - 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]).
 
 ### Changed
 
@@ -82,6 +83,7 @@ All notable changes to this project will be documented in this file.
 [#1022]: https://github.com/stackabletech/kafka-operator/pull/1022
 [#1024]: https://github.com/stackabletech/kafka-operator/pull/1024
 [#1029]: https://github.com/stackabletech/kafka-operator/pull/1029
+[#1035]: https://github.com/stackabletech/kafka-operator/pull/1035
 
 ## [26.7.0] - 2026-07-21
 

docs/modules/kafka/pages/usage-guide/operations/pod-placement.adoc

@@ -20,3 +20,23 @@ 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.
+
+[source,yaml]
+----
+affinity:
+  podAffinity:
+    preferredDuringSchedulingIgnoredDuringExecution:
+    - podAffinityTerm:
+        labelSelector:
+          matchLabels:
+            app.kubernetes.io/component: server
+            app.kubernetes.io/instance: opa-cluster-name
+            app.kubernetes.io/name: opa
+        topologyKey: kubernetes.io/hostname
+      weight: 50
+----
+
+`opa-cluster-name` is the `configMapName` from `spec.clusterConfig.authorization.opa`, which by convention is the name of the OpaCluster.

rust/operator-binary/src/controller/validate.rs

@@ -260,7 +260,11 @@ pub fn validate(
     let broker_role = &kafka.spec.brokers;
     let broker_groups = validate_role_group_configs(
         broker_role,
-        BrokerConfig::default_config(&kafka.name_any(), &KafkaRole::Broker.to_string()),
+        BrokerConfig::default_config(
+            &kafka.name_any(),
+            &KafkaRole::Broker.to_string(),
+            kafka.spec.cluster_config.authorization.opa.as_ref(),
+        ),
         cluster_id,
         AnyConfig::Broker,
         AnyConfigOverrides::Broker,

rust/operator-binary/src/crd/affinity.rs

@@ -1,13 +1,35 @@
 use stackable_operator::{
-    commons::affinity::{StackableAffinityFragment, affinity_between_role_pods},
-    k8s_openapi::api::core::v1::PodAntiAffinity,
+    commons::{
+        affinity::{StackableAffinityFragment, affinity_between_role_pods},
+        opa::OpaConfig,
+    },
+    k8s_openapi::api::core::v1::{PodAffinity, PodAntiAffinity},
 };
 
 use crate::crd::APP_NAME;
 
-pub fn get_affinity(cluster_name: &str, role: &str) -> StackableAffinityFragment {
+/// `opa_config` is only passed for roles that send authorization requests to OPA.
+pub fn get_affinity(
+    cluster_name: &str,
+    role: &str,
+    opa_config: Option<&OpaConfig>,
+) -> StackableAffinityFragment {
+    // With OPA authorization configured, the role sends its authorization requests to OPA, so
+    // prefer to place it next to the OPA Pods.
+    let pod_affinity = opa_config.map(|opa_config| PodAffinity {
+        preferred_during_scheduling_ignored_during_execution: Some(vec![
+            affinity_between_role_pods(
+                "opa",
+                &opa_config.config_map_name, // The discovery cm has the same name as the OpaCluster itself
+                "server",
+                50,
+            ),
+        ]),
+        required_during_scheduling_ignored_during_execution: None,
+    });
+
     StackableAffinityFragment {
-        pod_affinity: None,
+        pod_affinity,
         pod_anti_affinity: Some(PodAntiAffinity {
             preferred_during_scheduling_ignored_during_execution: Some(vec![
                 affinity_between_role_pods(APP_NAME, cluster_name, role, 70),
@@ -27,7 +49,9 @@ mod tests {
     use stackable_operator::{
         commons::affinity::StackableAffinity,
         k8s_openapi::{
-            api::core::v1::{PodAffinityTerm, PodAntiAffinity, WeightedPodAffinityTerm},
+            api::core::v1::{
+                PodAffinity, PodAffinityTerm, PodAntiAffinity, WeightedPodAffinityTerm,
+            },
             apimachinery::pkg::apis::meta::v1::LabelSelector,
         },
     };
@@ -39,6 +63,7 @@ mod tests {
 
     #[rstest]
     #[case(KafkaRole::Broker)]
+    #[case(KafkaRole::Controller)]
     fn test_affinity_defaults(#[case] role: KafkaRole) {
         let input = r#"
         apiVersion: kafka.stackable.tech/v1alpha1
@@ -52,10 +77,18 @@ mod tests {
             productVersion: 3.9.2
           clusterConfig:
             zookeeperConfigMapName: xyz
+            authorization:
+              opa:
+                configMapName: simple-opa
+                package: kafka
           brokers:
             roleGroups:
               default:
                 replicas: 1
+          controllers:
+            roleGroups:
+              default:
+                replicas: 1
         "#;
 
         let kafka = minimal_kafka(input);
@@ -70,7 +103,39 @@ mod tests {
         assert_eq!(
             merged_config.affinity,
             StackableAffinity {
-                pod_affinity: None,
+                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_anti_affinity: Some(PodAntiAffinity {
                     preferred_during_scheduling_ignored_during_execution: Some(vec![
                         WeightedPodAffinityTerm {
@@ -85,7 +150,7 @@ mod tests {
                                         ),
                                         (
                                             "app.kubernetes.io/component".to_string(),
-                                            "broker".to_string(),
+                                            role.to_string(),
                                         )
                                     ]))
                                 }),

rust/operator-binary/src/crd/role/broker.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,
@@ -87,9 +90,13 @@ pub struct BrokerConfig {
 }
 
 impl BrokerConfig {
-    pub fn default_config(cluster_name: &str, role: &str) -> BrokerConfigFragment {
+    pub fn default_config(
+        cluster_name: &str,
+        role: &str,
+        opa_config: Option<&OpaConfig>,
+    ) -> BrokerConfigFragment {
         BrokerConfigFragment {
-            common_config: CommonConfig::default_config(cluster_name, role),
+            common_config: CommonConfig::default_config(cluster_name, role, opa_config),
             bootstrap_listener_class: Some(DEFAULT_LISTENER_CLASS.clone()),
             broker_listener_class: Some(DEFAULT_LISTENER_CLASS.clone()),
             logging: product_logging::spec::default_logging(),

rust/operator-binary/src/crd/role/commons.rs

@@ -1,6 +1,6 @@
 use serde::{Deserialize, Serialize};
 use stackable_operator::{
-    commons::{affinity::StackableAffinity, resources::PvcConfig},
+    commons::{affinity::StackableAffinity, opa::OpaConfig, resources::PvcConfig},
     config::{fragment::Fragment, merge::Merge},
     k8s_openapi::api::core::v1::PersistentVolumeClaim,
     schemars::{self, JsonSchema},
@@ -70,9 +70,13 @@ impl CommonConfig {
     // Auto TLS certificate lifetime
     const DEFAULT_SECRET_LIFETIME: Duration = Duration::from_days_unchecked(1);
 
-    pub fn default_config(cluster_name: &str, role: &str) -> CommonConfigFragment {
+    pub fn default_config(
+        cluster_name: &str,
+        role: &str,
+        opa_config: Option<&OpaConfig>,
+    ) -> CommonConfigFragment {
         CommonConfigFragment {
-            affinity: get_affinity(cluster_name, role),
+            affinity: get_affinity(cluster_name, role, opa_config),
             graceful_shutdown_timeout: Some(Self::DEFAULT_GRACEFUL_SHUTDOWN_TIMEOUT),
             requested_secret_lifetime: Some(Self::DEFAULT_SECRET_LIFETIME),
         }

rust/operator-binary/src/crd/role/controller.rs

@@ -80,7 +80,9 @@ pub struct ControllerConfig {
 impl ControllerConfig {
     pub fn default_config(cluster_name: &str, role: &str) -> ControllerConfigFragment {
         ControllerConfigFragment {
-            common_config: CommonConfig::default_config(cluster_name, role),
+            // 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),
             logging: product_logging::spec::default_logging(),
             resources: ResourcesFragment {
                 cpu: CpuLimitsFragment {