fix: scale controller PDB maxUnavailable with replica count

CHANGELOG.md

@@ -16,6 +16,9 @@ All notable changes to this project will be documented in this file.
 - All product containers now run with `securityContext.runAsNonRoot` set to `true` to improve security ([#998]).
 - The reconciler now applies resources and derives the cluster status in discrete
   apply and update_status steps ([#1000]).
+- The controller role's PodDisruptionBudget `maxUnavailable` now scales with the number of
+  controller replicas (keeping a strict Raft quorum majority available) instead of being
+  hardcoded to `1` ([#1003]).
 
 ### Fixed
 
@@ -28,6 +31,7 @@ All notable changes to this project will be documented in this file.
 [#994]: https://github.com/stackabletech/kafka-operator/pull/994
 [#998]: https://github.com/stackabletech/kafka-operator/pull/998
 [#1000]: https://github.com/stackabletech/kafka-operator/pull/1000
+[#1003]: https://github.com/stackabletech/kafka-operator/pull/1003
 
 ## [26.7.0] - 2026-07-21
 

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

@@ -8,3 +8,8 @@ Unless you configure something else or disable the default PodDisruptionBudgets
 == Brokers
 Allow only a single Broker to be offline at any given time, regardless of the number of replicas or `roleGroups`.
 This is because we can not make any assumptions about topic replication factors.
+
+== Kraft Controllers
+The number of controllers that can be offline depends on the total number of replicas.
+The operator computes the `maxUnavailable` value using this formula `max(1, (replicas-1)/2)`.
+For more information, see the https://kafka.apache.org/42/operations/kraft/#deploying-considerations[Deployment Considerations] in the official Apache Kafka documentation.

rust/operator-binary/src/controller/build/resource/pdb.rs

@@ -1,3 +1,5 @@
+use std::cmp::max;
+
 use stackable_operator::{
     commons::pdb::PdbConfig, k8s_openapi::api::policy::v1::PodDisruptionBudget,
     v2::builder::pdb::pod_disruption_budget_builder_with_role,
@@ -19,7 +21,7 @@ pub fn build_pdb(
     }
     let max_unavailable = pdb.max_unavailable.unwrap_or(match role {
         KafkaRole::Broker => max_unavailable_brokers(),
-        KafkaRole::Controller => max_unavailable_controllers(),
+        KafkaRole::Controller => max_unavailable_controllers(controller_count(validated_cluster)),
     });
     let pdb = pod_disruption_budget_builder_with_role(
         validated_cluster,
@@ -34,12 +36,54 @@ pub fn build_pdb(
     Some(pdb)
 }
 
+/// Total number of controller replicas across all controller role groups.
+///
+/// Role groups without an explicit replica count (i.e. those left to a HorizontalPodAutoscaler)
+/// contribute nothing, as their size is not known at reconcile time.
+fn controller_count(cluster: &ValidatedCluster) -> u16 {
+    cluster
+        .role_group_configs
+        .get(&KafkaRole::Controller)
+        .into_iter()
+        .flat_map(|groups| groups.values())
+        .filter_map(|rg| rg.replicas)
+        .sum()
+}
+
 fn max_unavailable_brokers() -> u16 {
     // We can not make any assumptions about topic replication factors.
     1
 }
 
-fn max_unavailable_controllers() -> u16 {
-    // TODO: what do we want here?
-    1
+fn max_unavailable_controllers(num_controllers: u16) -> u16 {
+    // KRaft controllers form a Raft quorum: a strict majority must stay available for leader
+    // election and metadata commits to keep working, so at most `(N - 1) / 2` may be taken out
+    // at once.
+    let max_unavailable = num_controllers.saturating_sub(1) / 2;
+
+    // Clamp to at least a single controller allowed to be offline, so we don't block Kubernetes
+    // nodes from draining.
+    max(max_unavailable, 1)
+}
+
+#[cfg(test)]
+mod test {
+    use rstest::rstest;
+
+    use super::*;
+
+    #[rstest]
+    #[case(0, 1)]
+    #[case(1, 1)]
+    #[case(2, 1)]
+    #[case(3, 1)]
+    #[case(4, 1)]
+    #[case(5, 2)]
+    #[case(6, 2)]
+    #[case(7, 3)]
+    #[case(100, 49)]
+    fn test_max_unavailable_controllers(#[case] num_controllers: u16, #[case] expected: u16) {
+        let max_unavailable = max_unavailable_controllers(num_controllers);
+        assert_eq!(max_unavailable, expected);
+    }
 }