razvan · Nov 13, 2026 · Nov 13, 2026
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -39,7 +39,10 @@ All notable changes to this project will be documented in this file.
   On termination, a new `preStop` hook on the controller container (`kafka`) removes the pod from
   the voter list before shutdown.
   The controller `StatefulSet` is now scaled using `OrderedReady` instead of the `Parallel` strategy
-  to ensure only one voter is added/removed at a time and thus keep the quorum healthy ([#1010]).
+  to ensure only one voter is added/removed at a time and thus keep the quorum healthy. ([#1010]).
+- `controller.quorum.bootstrap.servers` always lists all currently running controller pod address.
+  It doesn't contain bootstrap service address anymore to avoid the problem of new controller pods
+  not being able to join the quorum because they connect to themselves ([#1063]).
 - Internal operator refactoring: introduce a build() step in the reconciler that
   assembles all relevant Kubernetes resources before anything is applied ([#985]).
 - Bump stackable-operator to 0.121.0 ([#994], [#1011], [#1021], [#1029], [#1043]).
@@ -124,6 +127,7 @@ All notable changes to this project will be documented in this file.
 [#1051]: https://github.com/stackabletech/kafka-operator/pull/1051
 [#1052]: https://github.com/stackabletech/kafka-operator/pull/1052
 [#1055]: https://github.com/stackabletech/kafka-operator/pull/1055
+[#1063]: https://github.com/stackabletech/kafka-operator/pull/1063
 
 ## [26.7.0] - 2026-07-21
 

diff --git a/docs/modules/kafka/pages/usage-guide/kraft-controller.adoc b/docs/modules/kafka/pages/usage-guide/kraft-controller.adoc
@@ -113,8 +113,10 @@ KRaft mode requires major configuration changes compared to ZooKeeper mode:
   with `kafka-storage.sh format --standalone`, bootstrapping a single-node quorum by itself.
   Every other controller formats with `--no-initial-controllers` and joins purely through the sidecar's `add-controller` call.
   Brokers always format with `--no-initial-controllers` too; they are never voters.
-* `controller.quorum.bootstrap.servers` points at each controller role group's own headless Service DNS name, not
-  individual pod addresses, unless Kerberos is enabled. See below for details related to Kerberos.
+* `controller.quorum.bootstrap.servers` lists the individual pod FQDNs of all controllers, not the role groups' headless Services.
+  A headless Service also resolves to pods that are not ready yet, so a starting controller could resolve it to its own address,
+  and then never find the quorum leader. With one entry per pod, Kafka picks a random entry for every attempt.
+  As a consequence, whenever the number of controller replicas changes, **both** the controller **and** broker pods are rolled by the restart controller.
 
 == Kerberos
 
@@ -135,11 +137,9 @@ When Kerberos is enabled:
   to initiate GSSAPI connections to its peers, not only accept them.
 * The `quorum-manager` sidecar and the `preStop` hook authenticate to the CONTROLLER listener
   as the controller's own pod principal.
-* `controller.quorum.bootstrap.servers` lists the individual pod FQDNs of all controllers.
-  This is required by Kerberos: a GSSAPI client derives the service principal from the hostname
+* `controller.quorum.bootstrap.servers` lists the individual pod FQDNs of all controllers (see above).
+  Kerberos requires this as well: a GSSAPI client derives the service principal from the hostname
   it dials, and the CONTROLLER listener can only offer the pod's own principal.
-  It means that, with Kerberos enabled, whenever the controller replicas is changed,
-  **both** the controller **and** broker pods are rolled by the restart controller.
 
 == Known Issues
 

diff --git a/rust/operator-binary/src/controller/build/properties/broker_properties.rs b/rust/operator-binary/src/controller/build/properties/broker_properties.rs
@@ -26,10 +26,7 @@ pub fn build(
     pod_descriptors: &[KafkaPodDescriptor],
     overrides: BTreeMap<String, String>,
 ) -> BTreeMap<String, String> {
-    let kraft_controllers = kraft_controllers(
-        pod_descriptors,
-        cluster_config.kafka_security.has_kerberos_enabled(),
-    );
+    let kraft_controllers = kraft_controllers(pod_descriptors);
 
     let mut result = BTreeMap::from([
         (

diff --git a/rust/operator-binary/src/controller/build/properties/controller_properties.rs b/rust/operator-binary/src/controller/build/properties/controller_properties.rs
@@ -25,11 +25,7 @@ pub fn build(
     pod_descriptors: &[KafkaPodDescriptor],
     overrides: BTreeMap<String, String>,
 ) -> BTreeMap<String, String> {
-    let kraft_controllers = kraft_controllers(
-        pod_descriptors,
-        cluster_config.kafka_security.has_kerberos_enabled(),
-    )
-    .join(",");
+    let kraft_controllers = kraft_controllers(pod_descriptors).join(",");
 
     let mut result = BTreeMap::from([
         (

diff --git a/rust/operator-binary/src/controller/build/properties/mod.rs b/rust/operator-binary/src/controller/build/properties/mod.rs
@@ -54,41 +54,35 @@ pub fn uses_legacy_log4j(product_version: &str) -> bool {
     product_version.starts_with("3.")
 }
 
-/// Builds the contents of the `controller.quorum.bootstrap.servers` property.
+/// Builds the contents of the `controller.quorum.bootstrap.servers` property: the FQDNs of all
+/// KRaft controller pods.
 ///
-/// When Kerberos is enabled, the list contains the FQDN pod names of the KRaft controllers.
+/// # Reasons for NOT using the role group's headless service
 ///
-/// For non-kerberized clusters, this list contains the headless service names
-/// of controller role groups.
+/// * When a new controller pod starts (without metadata) it finds the quorum leader
+///   by fetching from a bootstrap server. Kafka treats every entry of this list as
+///   one bootstrap node and picks a random entry for each attempt.
+///   The headless Service publishes not-ready addresses, so a single Service entry
+///   can resolve to the starting pod itself. That connection succeeds, so
+///   Kafka NEVER tries another address and the pod never learns the leader (see
+///   <https://github.com/stackabletech/kafka-operator/issues/1062>).
+/// * With Kerberos, the CONTROLLER listener's acceptor offers the principal `kafka/<pod-fqdn>`,
+///   which is also what the Raft voter endpoints advertise. Dialling the headless Service asks for
+///   `kafka/<service>` and fails authentication.
 ///
-/// # Why pod FQDNs under Kerberos
-///
-/// The CONTROLLER listener's acceptor offers the principal `kafka/<pod-fqdn>`, which
-/// is also what the Raft voter endpoints advertise.
-/// When a GSSAPI client uses the headless Service to ask for `kafka/<service>`,
-/// the authentication fails for every peer, so Kerberos-enabled clusters must list
-/// individual pod FQDNs.
-///
-/// That has a cost: the pod-FQDN list changes on every scaling operation (replica count change),
-/// so scaling a controller role group rolls *all* controller pods.
-/// Non-Kerberos clusters don't need pod-level addressing, so they keep the headless
-/// Service form and avoid that churn.
-pub(crate) fn kraft_controllers(
-    pod_descriptors: &[KafkaPodDescriptor],
-    kerberos_enabled: bool,
-) -> Vec<String> {
-    let controllers = pod_descriptors
+/// This has a cost: the list changes on every scaling operation of a controller role group, so
+/// scaling rolls all controller and broker pods. With OPA, `super.users` lists every pod as well,
+/// so those clusters already roll on any scaling operation.
+pub(crate) fn kraft_controllers(pod_descriptors: &[KafkaPodDescriptor]) -> Vec<String> {
+    pod_descriptors
         .iter()
-        .filter(|pd| pd.role == KafkaRole::Controller);
-
-    controllers
+        .filter(|pd| pd.role == KafkaRole::Controller)
         .map(|desc| {
-            let host = if kerberos_enabled {
-                desc.pod_fqdn()
-            } else {
-                desc.service_fqdn()
-            };
-            format!("{host}:{port}", port = desc.client_port)
+            format!(
+                "{host}:{port}",
+                host = desc.pod_fqdn(),
+                port = desc.client_port
+            )
         })
         .collect::<BTreeSet<_>>()
         .into_iter()
@@ -252,31 +246,7 @@ mod tests {
     }
 
     #[test]
-    fn kraft_controllers_lists_individual_pod_fqdns_under_kerberos() {
-        let pod_descriptors = vec![
-            pod_descriptor(KafkaRole::Controller, 0, 9093),
-            pod_descriptor(KafkaRole::Controller, 1, 9093),
-            pod_descriptor(KafkaRole::Controller, 2, 9093),
-            // Brokers must be filtered out of the controller quorum bootstrap servers list.
-            pod_descriptor(KafkaRole::Broker, 0, 9092),
-        ];
-
-        let quorum_bootstrap_servers = kraft_controllers(&pod_descriptors, true).join(",");
-
-        // Individual pod FQDNs, *not* the role group's headless Service. Under Kerberos the
-        // GSSAPI service principal is derived from the hostname the peer dials, and the
-        // CONTROLLER listener's acceptor can only offer one SPN -- the pod's own. Dialling
-        // the headless Service asks for `kafka/<service>` instead and is rejected.
-        assert_eq!(
-            quorum_bootstrap_servers,
-            "kafka-controller-default-0.kafka-controller-default-headless.default.svc.cluster.local:9093,\
-             kafka-controller-default-1.kafka-controller-default-headless.default.svc.cluster.local:9093,\
-             kafka-controller-default-2.kafka-controller-default-headless.default.svc.cluster.local:9093"
-        );
-    }
-
-    #[test]
-    fn kraft_controllers_lists_headless_services_without_kerberos() {
+    fn kraft_controllers_lists_individual_controller_pod_fqdns() {
         let mut other_group_pod = pod_descriptor(KafkaRole::Controller, 0, 9093);
         other_group_pod.role_group_statefulset_name = "kafka-controller-other"
             .parse()
@@ -285,73 +255,43 @@ mod tests {
             .parse()
             .expect("valid service name");
 
-        // Two replicas of the *default* role group and one of an *other* role group - the
-        // default group's Service entry must appear only once, regardless of replica count.
         let pod_descriptors = vec![
-            pod_descriptor(KafkaRole::Controller, 0, 9093),
             pod_descriptor(KafkaRole::Controller, 1, 9093),
+            pod_descriptor(KafkaRole::Controller, 0, 9093),
             other_group_pod,
             // Brokers must be filtered out of the controller quorum bootstrap servers list.
             pod_descriptor(KafkaRole::Broker, 0, 9092),
         ];
 
-        let quorum_bootstrap_servers = kraft_controllers(&pod_descriptors, false);
-
+        // One entry per controller pod, *not* the role group's headless Service: a single
+        // Service entry can resolve to the starting pod itself, which then never finds the
+        // leader (#1062). Under Kerberos the pod FQDN is also the only acceptable GSSAPI
+        // service principal.
         assert_eq!(
-            quorum_bootstrap_servers,
+            kraft_controllers(&pod_descriptors),
             vec![
-                "kafka-controller-default-headless.default.svc.cluster.local:9093".to_string(),
-                "kafka-controller-other-headless.default.svc.cluster.local:9093".to_string(),
+                "kafka-controller-default-0.kafka-controller-default-headless.default.svc.cluster.local:9093"
+                    .to_string(),
+                "kafka-controller-default-1.kafka-controller-default-headless.default.svc.cluster.local:9093"
+                    .to_string(),
+                "kafka-controller-other-0.kafka-controller-other-headless.default.svc.cluster.local:9093"
+                    .to_string(),
             ]
         );
     }
 
     #[test]
-    fn kraft_controllers_grows_with_the_replica_count_under_kerberos() {
-        let three_replicas = vec![
-            pod_descriptor(KafkaRole::Controller, 0, 9093),
-            pod_descriptor(KafkaRole::Controller, 1, 9093),
-            pod_descriptor(KafkaRole::Controller, 2, 9093),
-        ];
-        let five_replicas = vec![
-            pod_descriptor(KafkaRole::Controller, 0, 9093),
-            pod_descriptor(KafkaRole::Controller, 1, 9093),
-            pod_descriptor(KafkaRole::Controller, 2, 9093),
-            pod_descriptor(KafkaRole::Controller, 3, 9093),
-            pod_descriptor(KafkaRole::Controller, 4, 9093),
-        ];
-
-        // Deliberate consequence of per-pod addressing: unlike the headless-Service form used
-        // without Kerberos, this list changes with the replica count, so scaling a controller
-        // role group rolls the controller pods.
-        assert_eq!(kraft_controllers(&three_replicas, true).len(), 3);
-        assert_eq!(kraft_controllers(&five_replicas, true).len(), 5);
-        assert_ne!(
-            kraft_controllers(&three_replicas, true),
-            kraft_controllers(&five_replicas, true)
-        );
-    }
-
-    #[test]
-    fn kraft_controllers_is_stable_across_replica_count_changes_without_kerberos() {
-        let three_replicas = vec![
-            pod_descriptor(KafkaRole::Controller, 0, 9093),
-            pod_descriptor(KafkaRole::Controller, 1, 9093),
-            pod_descriptor(KafkaRole::Controller, 2, 9093),
-        ];
-        let five_replicas = vec![
-            pod_descriptor(KafkaRole::Controller, 0, 9093),
-            pod_descriptor(KafkaRole::Controller, 1, 9093),
-            pod_descriptor(KafkaRole::Controller, 2, 9093),
-            pod_descriptor(KafkaRole::Controller, 3, 9093),
-            pod_descriptor(KafkaRole::Controller, 4, 9093),
-        ];
+    fn kraft_controllers_grows_with_the_replica_count() {
+        let three_replicas: Vec<_> = (0..3)
+            .map(|replica| pod_descriptor(KafkaRole::Controller, replica, 9093))
+            .collect();
+        let five_replicas: Vec<_> = (0..5)
+            .map(|replica| pod_descriptor(KafkaRole::Controller, replica, 9093))
+            .collect();
 
-        // Without Kerberos, only adding or removing a whole role group changes this list, so
-        // scaling replicas within a role group does not roll the controller pods.
-        assert_eq!(
-            kraft_controllers(&three_replicas, false),
-            kraft_controllers(&five_replicas, false)
-        );
+        // Deliberate consequence of per-pod addressing: the list changes with the replica count,
+        // so scaling a controller role group rolls the controller and broker pods.
+        assert_eq!(kraft_controllers(&three_replicas).len(), 3);
+        assert_eq!(kraft_controllers(&five_replicas).len(), 5);
     }
 }