From cd51eab0a0249d8c4d8055e5e875d01b982e0c66 Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Fri, 13 Nov 2026 11:52:54 +0100 Subject: [PATCH 1/2] fix: persist KRaft bootstrap node id in cluster status --- CHANGELOG.md | 5 +- .../pages/usage-guide/kraft-controller.adoc | 8 +- extra/crds.yaml | 10 + rust/operator-binary/src/controller.rs | 195 ++++++++++++------ .../src/controller/build/command.rs | 96 ++++----- .../src/controller/build/mod.rs | 19 ++ .../controller/build/role_group/controller.rs | 6 +- .../src/controller/dereference.rs | 55 ++++- .../src/controller/update_status.rs | 33 +++ .../src/controller/validate.rs | 119 ++++++++++- rust/operator-binary/src/crd/mod.rs | 7 + .../75-add-controller-role-group.yaml.j2 | 47 +++++ .../kuttl/operations-kraft/75-assert.yaml.j2 | 48 +++++ 13 files changed, 518 insertions(+), 130 deletions(-) create mode 100644 tests/templates/kuttl/operations-kraft/75-add-controller-role-group.yaml.j2 create mode 100644 tests/templates/kuttl/operations-kraft/75-assert.yaml.j2 diff --git a/CHANGELOG.md b/CHANGELOG.md index b17eabd9e..d4cb81e1c 100644 --- 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. + The controller that bootstraps the quorum is chosen once and recorded in + `status.kraftBootstrapControllerNodeId`, so adding a controller role group never bootstraps a + second quorum ([#1010], [#XXXX]). - 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]). diff --git a/docs/modules/kafka/pages/usage-guide/kraft-controller.adoc b/docs/modules/kafka/pages/usage-guide/kraft-controller.adoc index 100498d44..a84602c55 100644 --- a/docs/modules/kafka/pages/usage-guide/kraft-controller.adoc +++ b/docs/modules/kafka/pages/usage-guide/kraft-controller.adoc @@ -109,8 +109,12 @@ KRaft mode requires major configuration changes compared to ZooKeeper mode: * Each controller pod runs an additional `quorum-manager` sidecar container that, on startup, admits the pod into the KRaft voter set. On pod restart or termination, the controller is removed from the voter set by the `preStop` hook on the `kafka` container. -* Exactly one controller (the one with the numerically lowest `node.id` among all controller pod descriptors) formats - with `kafka-storage.sh format --standalone`, bootstrapping a single-node quorum by itself. +* Exactly one controller formats with `kafka-storage.sh format --standalone`, bootstrapping a single-node quorum by itself. + It is the controller with the numerically lowest `node.id` when the cluster is first reconciled. + The operator records this choice in `status.kraftBootstrapControllerNodeId` of the KafkaCluster and never changes it afterwards, + so controller role groups added later (even with lower ``node.id``s) always join the existing quorum. + If the status is lost (e.g. the KafkaCluster is restored from a backup without its status), the operator chooses the bootstrap + controller again from the current spec. This is harmless as long as the PersistentVolumeClaims of the controllers are restored too. 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 diff --git a/extra/crds.yaml b/extra/crds.yaml index e7d5c7195..1f967db75 100644 --- a/extra/crds.yaml +++ b/extra/crds.yaml @@ -2238,6 +2238,16 @@ spec: - type type: object type: array + kraftBootstrapControllerNodeId: + description: |- + KRaft only: the `node.id` of the controller that bootstrapped the metadata quorum. + Chosen once, as the lowest controller `node.id` at the first reconcile, and never + changed afterwards, so that controller role groups added later never bootstrap a + second quorum, even if their `node.id`s are lower. + format: uint32 + minimum: 0.0 + nullable: true + type: integer type: object required: - spec diff --git a/rust/operator-binary/src/controller.rs b/rust/operator-binary/src/controller.rs index dcb5f74ad..8a934da9b 100644 --- a/rust/operator-binary/src/controller.rs +++ b/rust/operator-binary/src/controller.rs @@ -71,8 +71,10 @@ pub use stackable_operator::v2::types::operator::{RoleGroupName, RoleName}; use crate::{ controller::{ - apply::Applier, node_id_hasher::node_id_hash32_offset, security::ValidatedKafkaSecurity, - update_status::update_status, + apply::Applier, + node_id_hasher::node_id_hash32_offset, + security::ValidatedKafkaSecurity, + update_status::{persist_kraft_bootstrap_controller_node_id, update_status}, }, crd::{ APP_NAME, KAFKA_OPERATOR_NAME, KafkaPodDescriptor, MetadataManager, @@ -182,6 +184,10 @@ pub struct ValidatedCluster { /// addresses and the `Listener` watch triggers a new run once it does. pub bootstrap_listeners: Vec, pub agent_config: Option, + /// KRaft only: the `node.id` of the controller that bootstraps the metadata quorum. + /// Persisted once in the cluster status and then never changed. + /// Empty in ZooKeeper mode, or if no controller replica was ever requested. + pub kraft_bootstrap_controller_node_id: Option, } /// The number of pods a role group runs when its `replicas` is unset. @@ -196,6 +202,69 @@ pub(crate) fn replicas_or_default(replicas: Option) -> u16 { replicas.unwrap_or(DEFAULT_REPLICAS) } +/// The Kafka `node.id` of one predicted pod, see [`pod_node_ids`]. +#[derive(Clone, Debug, PartialEq)] +pub(crate) struct PodNodeId<'a> { + pub role: KafkaRole, + pub role_group_name: &'a RoleGroupName, + pub replica: u16, + pub node_id: u32, +} + +/// Predicts the `node.id` of every pod of the given role groups. +/// +/// A pod's `node.id` is the node-id hash offset of its role group plus its replica index. The +/// offsets must be unique across the whole cluster, so collisions are detected across all role +/// groups of both roles. +/// +/// Pods are predicted from the replica count rather than read from the live cluster to avoid +/// instance churn; an unset count falls back to [`DEFAULT_REPLICAS`]. +pub(crate) fn pod_node_ids<'a>( + broker_role_group_configs: &'a BTreeMap, + controller_role_group_configs: &'a BTreeMap, +) -> Result>, PodDescriptorsError> { + // In `KafkaRole` variant order, which decides the reported side of a hash collision. + let role_groups = broker_role_group_configs + .iter() + .map(|(role_group_name, rg)| (KafkaRole::Broker, role_group_name, rg.replicas)) + .chain( + controller_role_group_configs + .iter() + .map(|(role_group_name, rg)| (KafkaRole::Controller, role_group_name, rg.replicas)), + ); + + let mut pod_node_ids = Vec::new(); + let mut seen_hashes = HashMap::::new(); + + for (role, role_group_name, replicas) in role_groups { + let node_id_hash_offset = node_id_hash32_offset(&role, role_group_name.as_ref()); + + // The node id hash offset must be unique across the cluster. + if let Some((colliding_role, colliding_role_group)) = seen_hashes.get(&node_id_hash_offset) + { + return KafkaNodeIdHashCollisionSnafu { + role, + role_group: role_group_name.clone(), + colliding_role: colliding_role.clone(), + colliding_role_group: colliding_role_group.clone(), + } + .fail(); + } + seen_hashes.insert(node_id_hash_offset, (role.clone(), role_group_name.clone())); + + for replica in 0..replicas_or_default(replicas) { + pod_node_ids.push(PodNodeId { + role: role.clone(), + role_group_name, + replica, + node_id: node_id_hash_offset + u32::from(replica), + }); + } + } + + Ok(pod_node_ids) +} + impl ValidatedCluster { /// The `ObjectMeta` that lets [`ValidatedCluster`] act as the owner [`Resource`] for child /// objects: just `name`, `namespace` and `uid`. @@ -234,65 +303,27 @@ impl ValidatedCluster { requested_kafka_role: Option<&KafkaRole>, ) -> Result, PodDescriptorsError> { let client_port = self.cluster_config.kafka_security.client_port(); - let mut pod_descriptors = Vec::new(); - let mut seen_hashes = HashMap::::new(); - // In `KafkaRole` variant order, which decides the reported side of a hash collision. - let broker_role_groups: Vec<(&RoleGroupName, Option)> = self - .broker_role_group_configs - .iter() - .map(|(role_group_name, rg)| (role_group_name, rg.replicas)) - .collect(); - let controller_role_groups: Vec<(&RoleGroupName, Option)> = self - .controller_role_group_configs - .iter() - .map(|(role_group_name, rg)| (role_group_name, rg.replicas)) - .collect(); - let role_groups = [ - (KafkaRole::Broker, broker_role_groups), - (KafkaRole::Controller, controller_role_groups), - ]; - - for (role, role_groups) in &role_groups { - for (role_group_name, replicas) in role_groups { - let role_group_name = *role_group_name; - let node_id_hash_offset = node_id_hash32_offset(role, role_group_name.as_ref()); - - // The node id hash offset must be unique across the cluster. - if let Some((colliding_role, colliding_role_group)) = - seen_hashes.get(&node_id_hash_offset) - { - return KafkaNodeIdHashCollisionSnafu { - role: role.clone(), - role_group: role_group_name.clone(), - colliding_role: colliding_role.clone(), - colliding_role_group: colliding_role_group.clone(), - } - .fail(); - } - seen_hashes.insert(node_id_hash_offset, (role.clone(), role_group_name.clone())); - - if requested_kafka_role.is_none() || Some(role) == requested_kafka_role { - let resource_names = self.role_group_resource_names(role, role_group_name); - let role_group_statefulset_name = resource_names.stateful_set_name(); - let role_group_service_name = resource_names.headless_service_name(); - // Pods must be predicted from a concrete count (e.g. for KRaft quorum - // voters), so an unset replica count falls back to the default. - for replica in 0..replicas_or_default(*replicas) { - pod_descriptors.push(KafkaPodDescriptor { - namespace: self.namespace.clone(), - role_group_service_name: role_group_service_name.clone(), - role_group_statefulset_name: role_group_statefulset_name.clone(), - replica, - cluster_domain: self.cluster_domain.clone(), - node_id: node_id_hash_offset + u32::from(replica), - role: role.clone(), - client_port: client_port.clone(), - }); - } - } + let pod_descriptors = pod_node_ids( + &self.broker_role_group_configs, + &self.controller_role_group_configs, + )? + .into_iter() + .filter(|pod| requested_kafka_role.is_none_or(|requested| *requested == pod.role)) + .map(|pod| { + let resource_names = self.role_group_resource_names(&pod.role, pod.role_group_name); + KafkaPodDescriptor { + namespace: self.namespace.clone(), + role_group_service_name: resource_names.headless_service_name(), + role_group_statefulset_name: resource_names.stateful_set_name(), + replica: pod.replica, + cluster_domain: self.cluster_domain.clone(), + node_id: pod.node_id, + role: pod.role, + client_port: client_port.clone(), } - } + }) + .collect(); Ok(pod_descriptors) } @@ -572,6 +603,9 @@ pub async fn reconcile_kafka( .await .context(DereferenceSnafu)?; + let persisted_kraft_bootstrap_controller_node_id = + dereferenced_objects.persisted_kraft_bootstrap_controller_node_id; + // validate (no client required) let validated_cluster = validate::validate( kafka, @@ -589,6 +623,15 @@ pub async fn reconcile_kafka( "The following security settings are used" ); + // Persist a newly chosen KRaft bootstrap controller before any controller Pod can act on it. + if let Some(node_id) = validated_cluster.kraft_bootstrap_controller_node_id + && persisted_kraft_bootstrap_controller_node_id != Some(node_id) + { + persist_kraft_bootstrap_controller_node_id(client, kafka, node_id) + .await + .context(UpdateStatusSnafu)?; + } + // build (no client required) let resources = build::build(&validated_cluster).context(BuildResourcesSnafu)?; @@ -605,9 +648,14 @@ pub async fn reconcile_kafka( .context(ApplyResourcesSnafu)?; // update status (client required) - update_status(client, kafka, applied) - .await - .context(UpdateStatusSnafu)?; + update_status( + client, + kafka, + validated_cluster.kraft_bootstrap_controller_node_id, + applied, + ) + .await + .context(UpdateStatusSnafu)?; Ok(Action::await_change()) } @@ -864,6 +912,7 @@ pub(crate) mod test_support { authorization_config, kubernetes_cluster_info: cluster_info(), bootstrap_listeners: Vec::new(), + persisted_kraft_bootstrap_controller_node_id: None, }, &operator_environment(), Some(OPERATOR_IMAGE), @@ -890,11 +939,33 @@ pub(crate) mod test_support { authorization_config: None, kubernetes_cluster_info: cluster_info(), bootstrap_listeners: Vec::new(), + persisted_kraft_bootstrap_controller_node_id: None, }, &operator_environment(), operator_image, ) } + + /// Runs the real validate step as if a previous reconcile had persisted the given KRaft + /// bootstrap controller `node.id` in the cluster status. + pub fn validated_cluster_with_persisted_kraft_bootstrap( + kafka: &v1alpha1::KafkaCluster, + persisted_kraft_bootstrap_controller_node_id: Option, + ) -> ValidatedCluster { + validate( + kafka, + DereferencedObjects { + authentication_classes: ResolvedAuthenticationClasses::new(Vec::new()), + authorization_config: None, + kubernetes_cluster_info: cluster_info(), + bootstrap_listeners: Vec::new(), + persisted_kraft_bootstrap_controller_node_id, + }, + &operator_environment(), + Some(OPERATOR_IMAGE), + ) + .expect("validate should succeed for the test fixture") + } } #[cfg(test)] diff --git a/rust/operator-binary/src/controller/build/command.rs b/rust/operator-binary/src/controller/build/command.rs index 63e6d489e..e6a7553c4 100644 --- a/rust/operator-binary/src/controller/build/command.rs +++ b/rust/operator-binary/src/controller/build/command.rs @@ -15,8 +15,8 @@ use super::{container::FILE_LOG_DIRECTORY, properties::ConfigFileName}; use crate::{ controller::{build::security::copy_opa_tls_cert_command, security::ValidatedKafkaSecurity}, crd::{ - BROKER_ID_POD_MAP_DIR, KafkaPodDescriptor, METRICS_PORT, STACKABLE_CONFIG_DIR, - STACKABLE_KERBEROS_KRB5_PATH, STACKABLE_LOG_CONFIG_DIR, role::KafkaRole, + BROKER_ID_POD_MAP_DIR, METRICS_PORT, STACKABLE_CONFIG_DIR, STACKABLE_KERBEROS_KRB5_PATH, + STACKABLE_LOG_CONFIG_DIR, }, }; @@ -124,33 +124,31 @@ fn broker_start_command(kraft_mode: bool) -> String { /// Selects the quorum format flag for the given controller. /// -/// The controller with lowest `node_id` starts with `--standalone` while all others -/// start with `--no-initial-controllers` and are added later to the voter list -/// by the `quorum-manager`. +/// The bootstrap controller (chosen once and persisted, see +/// [`crate::crd::KafkaClusterStatus::kraft_bootstrap_controller_node_id`]) starts with `--standalone` while all +/// others start with `--no-initial-controllers` and are added later to the voter list by the +/// `quorum-manager`. Without a bootstrap controller, every controller uses +/// `--no-initial-controllers`. /// -/// Known limitation: If the controller with the lowest `node_id` loses it's PVC it will -/// create a new conflicting quorum upon restart. -fn controller_quorum_format_flag(controller_descriptors: &[KafkaPodDescriptor]) -> String { - let bootstrap_node_id = controller_descriptors - .iter() - .filter(|descriptor| descriptor.role == KafkaRole::Controller) - .map(|descriptor| descriptor.node_id) - .min() - .unwrap_or(0); - - formatdoc! {" - if [ \"$REPLICA_ID\" = \"{bootstrap_node_id}\" ]; then - FORMAT_QUORUM_FLAG=--standalone - else - FORMAT_QUORUM_FLAG=--no-initial-controllers - fi - " +/// Known limitation: If the bootstrap controller loses its PVC it will create a new conflicting +/// quorum upon restart. +fn controller_quorum_format_flag(bootstrap_node_id: Option) -> String { + match bootstrap_node_id { + Some(bootstrap_node_id) => formatdoc! {" + if [ \"$REPLICA_ID\" = \"{bootstrap_node_id}\" ]; then + FORMAT_QUORUM_FLAG=--standalone + else + FORMAT_QUORUM_FLAG=--no-initial-controllers + fi + " + }, + None => "FORMAT_QUORUM_FLAG=--no-initial-controllers\n".to_owned(), } } pub fn controller_kafka_container_command( kafka_security: &ValidatedKafkaSecurity, - controller_descriptors: Vec, + bootstrap_node_id: Option, ) -> String { formatdoc! {" {COMMON_BASH_TRAP_FUNCTIONS} @@ -194,7 +192,7 @@ pub fn controller_kafka_container_command( jaas_file = ConfigFileName::Jaas, admin_client_source = ADMIN_CLIENT_PROPERTIES_SOURCE_PATH, admin_client_config = ADMIN_CLIENT_PROPERTIES_PATH, - quorum_format_flag = controller_quorum_format_flag(&controller_descriptors), + quorum_format_flag = controller_quorum_format_flag(bootstrap_node_id), create_vector_shutdown_file_command = create_vector_shutdown_file_command(STACKABLE_LOG_DIR) } } @@ -405,7 +403,7 @@ mod tests { #[test] fn controller_command_does_not_export_a_realm_without_kerberos() { - let command = controller_kafka_container_command(&plaintext(), vec![]); + let command = controller_kafka_container_command(&plaintext(), None); assert!(!command.contains("KERBEROS_REALM")); } @@ -414,7 +412,7 @@ mod tests { // `jaas.properties` is always present in the ConfigMap (empty when Kerberos is off), // so the copy is unconditional, matching `broker_start_command`. for security in [kerberos(), plaintext()] { - let command = controller_kafka_container_command(&security, vec![]); + let command = controller_kafka_container_command(&security, None); assert!(command.contains("cp /stackable/config/jaas.properties /tmp/jaas.properties")); assert!(command.contains("config-utils template /tmp/jaas.properties")); } @@ -446,7 +444,7 @@ mod tests { #[test] fn controller_command_templates_the_admin_client_config_for_pre_stop() { - let command = controller_kafka_container_command(&kerberos(), vec![]); + let command = controller_kafka_container_command(&kerberos(), None); assert!( command.contains( "cp /stackable/config/admin-client.properties /tmp/admin-client.properties" @@ -1157,38 +1155,12 @@ mod tests { ); } - /// Builds a minimal [`KafkaPodDescriptor`] for the given role and replica. - fn pod_descriptor(role: KafkaRole, replica: u16, node_id: u32) -> KafkaPodDescriptor { - KafkaPodDescriptor { - namespace: "default".parse().expect("valid namespace name"), - role_group_statefulset_name: "kafka-controller-default" - .parse() - .expect("valid statefulset name"), - role_group_service_name: "kafka-controller-default-headless" - .parse() - .expect("valid service name"), - replica, - cluster_domain: stackable_operator::commons::networking::DomainName::try_from( - "cluster.local", - ) - .expect("valid domain"), - node_id, - role, - client_port: 9093.into(), - } - } - - /// The controller with the lowest `node_id` bootstraps the quorum by itself - /// (`--standalone`); every other controller joins via the `quorum-manager` sidecar's - /// `add-controller` loop (`--no-initial-controllers`). + /// The bootstrap controller bootstraps the quorum by itself (`--standalone`); every other + /// controller joins via the `quorum-manager` sidecar's `add-controller` loop + /// (`--no-initial-controllers`). #[test] - fn controller_kafka_container_command_branches_on_the_lowest_node_id() { - let descriptors = vec![ - pod_descriptor(KafkaRole::Controller, 0, 5), - pod_descriptor(KafkaRole::Controller, 1, 6), - pod_descriptor(KafkaRole::Controller, 2, 7), - ]; - let command = controller_kafka_container_command(&plaintext(), descriptors); + fn controller_kafka_container_command_branches_on_the_bootstrap_node_id() { + let command = controller_kafka_container_command(&plaintext(), Some(5)); assert!(command.contains(r#"if [ "$REPLICA_ID" = "5" ]; then"#)); assert!(command.contains("FORMAT_QUORUM_FLAG=--standalone")); @@ -1201,6 +1173,14 @@ mod tests { assert!(!command.contains("0000000000-")); } + /// Without a bootstrap controller, no controller may bootstrap a quorum. + #[test] + fn controller_kafka_container_command_without_bootstrap_node_id_never_uses_standalone() { + let command = controller_kafka_container_command(&plaintext(), None); + assert!(command.contains("FORMAT_QUORUM_FLAG=--no-initial-controllers")); + assert!(!command.contains("--standalone")); + } + /// Brokers are never voters and never the bootstrap candidate — they always join (or, for /// a fresh cluster, simply never assert any voter membership) via `--no-initial-controllers`. #[test] diff --git a/rust/operator-binary/src/controller/build/mod.rs b/rust/operator-binary/src/controller/build/mod.rs index 30664d8ab..83e8b1a47 100644 --- a/rust/operator-binary/src/controller/build/mod.rs +++ b/rust/operator-binary/src/controller/build/mod.rs @@ -530,6 +530,25 @@ mod tests { build(&cluster).expect("build succeeds when the whole KRaft cluster is stopped"); } + /// The KRaft bootstrap controller recorded on the [`ValidatedCluster`] (i.e. the persisted + /// one, if any) is what every controller's start command carries -- not the lowest + /// `node.id` of the current spec. + #[test] + fn build_uses_the_resolved_kraft_bootstrap_controller() { + let mut cluster = kraft_mode_cluster(); + cluster.kraft_bootstrap_controller_node_id = Some(42); + let resources = build(&cluster).expect("build succeeds"); + + let controller_stateful_set = resources + .stateful_sets + .iter() + .find(|sts| sts.metadata.name.as_deref() == Some("simple-kafka-controller-default")) + .expect("the controller StatefulSet is built"); + let controller_stateful_set = + serde_json::to_string(controller_stateful_set).expect("serializable StatefulSet"); + assert!(controller_stateful_set.contains(r#"[ \"$REPLICA_ID\" = \"42\" ]"#)); + } + #[test] fn build_produces_expected_resource_names() { let cluster = kraft_mode_cluster(); 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 3368dd11f..e45be5457 100644 --- a/rust/operator-binary/src/controller/build/role_group/controller.rs +++ b/rust/operator-binary/src/controller/build/role_group/controller.rs @@ -276,10 +276,6 @@ fn role_pod( .with_value(&KAFKA_NODE_ID_OFFSET, &node_id_offset) .merge(rg_config.env_overrides.clone()); - let controller_pod_descriptors = cluster - .pod_descriptors(Some(&ROLE)) - .context(BuildPodDescriptorsSnafu)?; - let startup_probe = probes::controller_tcp_probe(kafka_security.client_port(), 5, 5, 60) .context(BuildProbeSnafu)?; let liveness_probe = probes::controller_stuck_unattached_liveness_probe( @@ -304,7 +300,7 @@ fn role_pod( ]) .args(vec![controller_kafka_container_command( kafka_security, - controller_pod_descriptors, + cluster.kraft_bootstrap_controller_node_id, )]); cb_kafka diff --git a/rust/operator-binary/src/controller/dereference.rs b/rust/operator-binary/src/controller/dereference.rs index 605e34c79..876945679 100644 --- a/rust/operator-binary/src/controller/dereference.rs +++ b/rust/operator-binary/src/controller/dereference.rs @@ -23,7 +23,10 @@ use stackable_operator::{ utils::cluster_info::KubernetesClusterInfo, v2::{ controller_utils::{get_cluster_name, get_namespace}, - types::kubernetes::ListenerName, + types::{ + kubernetes::{ListenerName, NamespaceName}, + operator::ClusterName, + }, }, }; @@ -66,6 +69,11 @@ pub enum Error { source: stackable_operator::client::Error, listener_name: ListenerName, }, + + #[snafu(display("failed to fetch the live KafkaCluster"))] + FetchKafkaCluster { + source: stackable_operator::client::Error, + }, } type Result = std::result::Result; @@ -75,6 +83,10 @@ pub struct DereferencedObjects { pub authorization_config: Option, pub kubernetes_cluster_info: KubernetesClusterInfo, pub bootstrap_listeners: Vec, + /// KRaft only: the bootstrap controller `node.id` persisted in the cluster status by a + /// previous reconcile run, if any. See + /// [`crate::crd::KafkaClusterStatus::kraft_bootstrap_controller_node_id`]. + pub persisted_kraft_bootstrap_controller_node_id: Option, } pub async fn dereference( @@ -121,10 +133,51 @@ pub async fn dereference( } } + let persisted_kraft_bootstrap_controller_node_id = + persisted_kraft_bootstrap_controller_node_id(client, kafka, &cluster_name, &namespace) + .await?; + Ok(DereferencedObjects { authentication_classes, authorization_config, kubernetes_cluster_info: client.kubernetes_cluster_info.clone(), bootstrap_listeners, + persisted_kraft_bootstrap_controller_node_id, }) } + +/// Reads the persisted KRaft bootstrap controller `node.id` from the cluster status. +/// +/// The reconciled object comes from the controller's cache, which may not yet contain the status +/// patch written by the previous run (see `reconcile_kafka`). If the cached object carries no +/// value, the live object is fetched before concluding that none was persisted yet. Otherwise a +/// stale cache combined with a spec change (e.g. a newly added controller role group) could +/// choose a different bootstrap controller and start a second quorum. +async fn persisted_kraft_bootstrap_controller_node_id( + client: &Client, + kafka: &v1alpha1::KafkaCluster, + cluster_name: &ClusterName, + namespace: &NamespaceName, +) -> Result> { + fn from_status(kafka: &v1alpha1::KafkaCluster) -> Option { + kafka + .status + .as_ref() + .and_then(|status| status.kraft_bootstrap_controller_node_id) + } + + // Only KRaft clusters have controllers. + if kafka.spec.controllers.is_none() { + return Ok(None); + } + + if let Some(node_id) = from_status(kafka) { + return Ok(Some(node_id)); + } + + let live = client + .get_opt::(cluster_name.as_ref(), namespace.as_ref()) + .await + .context(FetchKafkaClusterSnafu)?; + Ok(live.as_ref().and_then(from_status)) +} diff --git a/rust/operator-binary/src/controller/update_status.rs b/rust/operator-binary/src/controller/update_status.rs index 3c24c96e6..71df78e54 100644 --- a/rust/operator-binary/src/controller/update_status.rs +++ b/rust/operator-binary/src/controller/update_status.rs @@ -22,6 +22,11 @@ pub enum Error { ApplyStatus { source: stackable_operator::client::Error, }, + + #[snafu(display("failed to persist the KRaft bootstrap controller node id in the status"))] + PersistKraftBootstrapControllerNodeId { + source: stackable_operator::client::Error, + }, } type Result = std::result::Result; @@ -29,9 +34,13 @@ type Result = std::result::Result; /// Computes the cluster status from the applied resources and patches it onto the /// [`v1alpha1::KafkaCluster`]. Takes [`KubernetesResources`] so the type system /// proves the status derives from applied resources, not merely built ones. +/// +/// `kraft_bootstrap_controller_node_id` must always be passed (if known): the status is written +/// with server-side apply, which removes fields this field manager wrote before but omits now. pub async fn update_status( client: &Client, kafka: &v1alpha1::KafkaCluster, + kraft_bootstrap_controller_node_id: Option, applied: KubernetesResources, ) -> Result<()> { let mut ss_cond_builder = StatefulSetConditionBuilder::default(); @@ -44,6 +53,7 @@ pub async fn update_status( let status = KafkaClusterStatus { conditions: compute_conditions(kafka, &[&ss_cond_builder, &cluster_operation_cond_builder]), + kraft_bootstrap_controller_node_id, }; client @@ -53,3 +63,26 @@ pub async fn update_status( Ok(()) } + +/// Persists a newly chosen KRaft bootstrap controller `node.id` in the cluster status. +/// +/// Must run before the StatefulSets are applied: once a controller Pod has formatted its storage +/// with `--standalone`, the choice must survive even if a later step of this reconcile run fails +/// and the spec changes before the next one. +/// +/// A JSON merge patch touches only this field, so the conditions are left alone. The subsequent +/// [`update_status`] writes the same value with server-side apply. +pub async fn persist_kraft_bootstrap_controller_node_id( + client: &Client, + kafka: &v1alpha1::KafkaCluster, + node_id: u32, +) -> Result<()> { + client + .merge_patch_status( + kafka, + &serde_json::json!({ "kraftBootstrapControllerNodeId": node_id }), + ) + .await + .context(PersistKraftBootstrapControllerNodeIdSnafu)?; + Ok(()) +} diff --git a/rust/operator-binary/src/controller/validate.rs b/rust/operator-binary/src/controller/validate.rs index 4ca0743c3..d8bc23542 100644 --- a/rust/operator-binary/src/controller/validate.rs +++ b/rust/operator-binary/src/controller/validate.rs @@ -34,7 +34,7 @@ use crate::{ ValidatedAgentConfig, ValidatedBrokerConfig, ValidatedCluster, ValidatedClusterConfig, ValidatedControllerConfig, ValidatedRoleConfig, dereference::DereferencedObjects, - replicas_or_default, + pod_node_ids, replicas_or_default, security::{self, ValidatedKafkaSecurity}, }, crd::{ @@ -71,6 +71,11 @@ pub enum Error { source: stackable_operator::config::fragment::ValidationError, }, + #[snafu(display("failed to predict the node ids of the pods"))] + PredictNodeIds { + source: crate::controller::PodDescriptorsError, + }, + #[snafu(display("invalid metadata manager"))] InvalidMetadataManager { source: crate::crd::Error }, @@ -393,6 +398,15 @@ pub fn validate( }) .transpose()?; + let kraft_bootstrap_controller_node_id = match metadata_manager { + crate::crd::MetadataManager::KRaft => resolve_kraft_bootstrap_controller_node_id( + dereferenced_objects.persisted_kraft_bootstrap_controller_node_id, + &broker_role_group_configs, + &controller_role_group_configs, + )?, + _ => None, + }; + Ok(ValidatedCluster { metadata: ValidatedCluster::object_meta(&name, &namespace, &uid), name, @@ -418,9 +432,33 @@ pub fn validate( controller_role_group_configs, bootstrap_listeners: dereferenced_objects.bootstrap_listeners, agent_config, + kraft_bootstrap_controller_node_id, }) } +/// Decides the KRaft bootstrap controller's `node.id`, i.e. the one controller that formats with +/// `--standalone`. +/// +/// A persisted choice always wins, even if the spec now contains a controller with a lower +/// `node.id`: a controller role group added later can hash to a lower node-id offset, and its +/// pod-0 (with a fresh PVC) would otherwise bootstrap a second, independent quorum under the same +/// cluster id. Only if nothing is persisted yet (a new cluster) is the lowest controller `node.id` +/// of the current spec chosen. Returns `None` if there is neither (no controller replicas). +fn resolve_kraft_bootstrap_controller_node_id( + persisted: Option, + broker_role_group_configs: &BTreeMap, + controller_role_group_configs: &BTreeMap, +) -> Result> { + let controller_node_ids = + pod_node_ids(broker_role_group_configs, controller_role_group_configs) + .context(PredictNodeIdsSnafu)? + .into_iter() + .filter(|pod| pod.role == KafkaRole::Controller) + .map(|pod| pod.node_id); + + Ok(persisted.or_else(|| controller_node_ids.into_iter().min())) +} + /// The role-level config, resolved from the role's `roleConfig`. fn validated_role_config(role_config: &GenericRoleConfig) -> ValidatedRoleConfig { // Destructured without `..`, so a field added upstream fails to compile here instead of @@ -677,6 +715,7 @@ mod tests { app_version_label, broker_role_group_config, client_auth_tls, controller_role_group_config, kerberos, minimal_kafka, plaintext, server_tls, validate_with_operator_image, validated_cluster, + validated_cluster_with_persisted_kraft_bootstrap, }, }, crd::v1alpha1::UserProvidedAuthentication, @@ -1149,4 +1188,82 @@ mod tests { let _cluster = validated_cluster(&kafka); } + + /// A KRaft cluster with the `default` controller role group plus a second one, `extra`, + /// whose node-id offset (87094686) is lower than `default`'s (2110489703). + fn kraft_cluster_with_lower_offset_controller_group() -> crate::crd::v1alpha1::KafkaCluster { + minimal_kafka( + r#" + apiVersion: kafka.stackable.tech/v1alpha1 + kind: KafkaCluster + metadata: + name: simple-kafka + namespace: default + uid: 12345678-1234-1234-1234-123456789012 + spec: + image: + productVersion: 4.3.1 + clusterConfig: + metadataManager: KRaft + controllers: + roleGroups: + default: + replicas: 3 + extra: + replicas: 1 + brokers: + roleGroups: + default: + replicas: 1 + "#, + ) + } + + /// Regression test: adding a controller role group that hashes to a lower node-id offset + /// must not move the bootstrap controller, or its fresh pod-0 bootstraps a second quorum. + #[test] + fn kraft_bootstrap_controller_stays_persisted_when_lower_offset_group_is_added() { + let cluster = validated_cluster_with_persisted_kraft_bootstrap( + &kraft_cluster_with_lower_offset_controller_group(), + Some(2110489703), + ); + assert_eq!(cluster.kraft_bootstrap_controller_node_id, Some(2110489703)); + } + + /// Without a persisted choice (a new cluster), the lowest controller `node.id` bootstraps. + #[test] + fn kraft_bootstrap_controller_is_lowest_node_id_for_new_cluster() { + let cluster = validated_cluster_with_persisted_kraft_bootstrap( + &kraft_cluster_with_lower_offset_controller_group(), + None, + ); + assert_eq!(cluster.kraft_bootstrap_controller_node_id, Some(87094686)); + } + + #[test] + fn kraft_bootstrap_controller_is_none_in_zookeeper_mode() { + let kafka = minimal_kafka( + r#" + apiVersion: kafka.stackable.tech/v1alpha1 + kind: KafkaCluster + metadata: + name: simple-kafka + namespace: default + uid: 12345678-1234-1234-1234-123456789012 + spec: + image: + productVersion: 3.9.2 + clusterConfig: + zookeeperConfigMapName: xyz + brokers: + roleGroups: + default: + replicas: 1 + "#, + ); + assert_eq!( + validated_cluster(&kafka).kraft_bootstrap_controller_node_id, + None + ); + } } diff --git a/rust/operator-binary/src/crd/mod.rs b/rust/operator-binary/src/crd/mod.rs index 5035c1213..e1bec5070 100644 --- a/rust/operator-binary/src/crd/mod.rs +++ b/rust/operator-binary/src/crd/mod.rs @@ -410,6 +410,13 @@ impl KafkaPodDescriptor { pub struct KafkaClusterStatus { #[serde(default)] pub conditions: Vec, + + /// KRaft only: the `node.id` of the controller that bootstrapped the metadata quorum. + /// Chosen once, as the lowest controller `node.id` at the first reconcile, and never + /// changed afterwards, so that controller role groups added later never bootstrap a + /// second quorum, even if their `node.id`s are lower. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub kraft_bootstrap_controller_node_id: Option, } #[derive( diff --git a/tests/templates/kuttl/operations-kraft/75-add-controller-role-group.yaml.j2 b/tests/templates/kuttl/operations-kraft/75-add-controller-role-group.yaml.j2 new file mode 100644 index 000000000..5a036e1b0 --- /dev/null +++ b/tests/templates/kuttl/operations-kraft/75-add-controller-role-group.yaml.j2 @@ -0,0 +1,47 @@ +# +# Regression test: adding a controller role group whose node ids are lower than those of the +# controller that bootstrapped the quorum must not bootstrap a second quorum. +# +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestStep +timeout: 600 +--- +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 %} + clusterConfig: + metadataManager: KRaft +{% if lookup('env', 'VECTOR_AGGREGATOR') %} + vectorAggregatorConfigMapName: vector-aggregator-discovery +{% endif %} + controllers: + config: + logging: + enableVectorAgent: {{ lookup('env', 'VECTOR_AGGREGATOR') | length > 0 }} + roleGroups: + default: + replicas: 3 + # `extra` hashes to a lower node-id offset (87094686) than `default` (2110489703). Its + # pod-0 must join the existing quorum, not bootstrap a second one. + extra: + replicas: 1 + brokers: + config: + logging: + enableVectorAgent: {{ lookup('env', 'VECTOR_AGGREGATOR') | length > 0 }} + roleGroups: + default: + replicas: 3 + clusterOperation: + stopped: false + reconciliationPaused: false diff --git a/tests/templates/kuttl/operations-kraft/75-assert.yaml.j2 b/tests/templates/kuttl/operations-kraft/75-assert.yaml.j2 new file mode 100644 index 000000000..6d9101466 --- /dev/null +++ b/tests/templates/kuttl/operations-kraft/75-assert.yaml.j2 @@ -0,0 +1,48 @@ +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestAssert +timeout: 600 +commands: + - script: kubectl -n $NAMESPACE wait --for=condition=available kafkaclusters.kafka.stackable.tech/test-kafka --timeout 301s + - script: | + # :9093 is the TLS client port of this test fixture's default security config, not a + # fixed Kafka port - if the fixture's TLS/port config changes, update this too. + # + # Ask both an original controller and the new `extra` controller: a second, rogue quorum + # would show up as `extra-0` leading a single-voter quorum of its own. + for controller in test-kafka-controller-default-0.test-kafka-controller-default-headless \ + test-kafka-controller-extra-0.test-kafka-controller-extra-headless; do + kubectl exec -n $NAMESPACE test-kafka-controller-default-0 -c kafka -- \ + /stackable/kafka/bin/kafka-metadata-quorum.sh \ + --bootstrap-controller $controller.$NAMESPACE.svc.cluster.local:9093 \ + --command-config /stackable/config/admin-client.properties \ + describe --replication | tail -n +2 | awk '$NF == "Leader" || $NF == "Follower"' | wc -l | grep -q '^4$' || exit 1 + done + # `timeout` is known-inert here: kuttl's TestAssert `commands` don't read this field (only + # TestStep commands do); left in place only as documentation of the intended budget. + timeout: 30 +--- +# The bootstrap controller is the one chosen when the cluster was created (default-0), not the +# lowest node id of the current spec (extra-0). +apiVersion: kafka.stackable.tech/v1alpha1 +kind: KafkaCluster +metadata: + name: test-kafka +status: + kraftBootstrapControllerNodeId: 2110489703 +--- +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-controller-extra +status: + readyReplicas: 1 + replicas: 1 From fa3a0f39f5adcf2f1e24e0a80ee3e7de18cbb905 Mon Sep 17 00:00:00 2001 From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com> Date: Fri, 13 Nov 2026 11:56:32 +0100 Subject: [PATCH 2/2] update changelog --- CHANGELOG.md | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d4cb81e1c..bfc463968 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -42,7 +42,7 @@ All notable changes to this project will be documented in this file. to ensure only one voter is added/removed at a time and thus keep the quorum healthy. The controller that bootstraps the quorum is chosen once and recorded in `status.kraftBootstrapControllerNodeId`, so adding a controller role group never bootstraps a - second quorum ([#1010], [#XXXX]). + second quorum ([#1010], [#1061]). - 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]). @@ -127,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 +[#1061]: https://github.com/stackabletech/kafka-operator/pull/1061 ## [26.7.0] - 2026-07-21