fix: Re-emit existing discovery CM if listener is not ready

CHANGELOG.md

@@ -28,8 +28,6 @@ All notable changes to this project will be documented in this file.
   Previously a ZookeeperCluster without it was accepted by the API server but failed reconciliation ([#1060]).
 - The reconciler now applies resources and derives the cluster status in discrete
   apply and update_status steps for the `zk_controller` and `znode_controller` ([#1069]).
-- The discovery ConfigMap is now always written, with empty `ZOOKEEPER` and `ZOOKEEPER_HOSTS` while
-  the listener publishes no addresses ([#1069]).
 - All product containers now run with `securityContext.runAsNonRoot` set to `true` to improve security ([#1070]).
 - Remove the `app.kubernetes.io/component` and `app.kubernetes.io/role-group` labels from the
   resources they don't apply to (previously set to `none` or the placeholder value `discovery`)
@@ -43,6 +41,8 @@ All notable changes to this project will be documented in this file.
 - All pods now set `securityContext.seccompProfile.type` to `RuntimeDefault` ([#1097]).
 - Replace the unmaintained `serde_yaml` with `yaml_serde` ([#1097]).
 - The operator Deployment runs with a restricted security context and a read-only root filesystem, so it can be deployed into namespaces that enforce the `restricted` Pod Security Standard ([#1101]).
+- An existing discovery ConfigMap is kept unchanged while the role listener has no address and a new one is only created once the address is available.
+  Previously, reconciliation failed while the address was missing ([#1069], [#1105]).
 
 ### Fixed
 
@@ -74,6 +74,7 @@ All notable changes to this project will be documented in this file.
 [#1100]: https://github.com/stackabletech/zookeeper-operator/pull/1100
 [#1101]: https://github.com/stackabletech/zookeeper-operator/pull/1101
 [#1102]: https://github.com/stackabletech/zookeeper-operator/pull/1102
+[#1105]: https://github.com/stackabletech/zookeeper-operator/pull/1105
 [#1106]: https://github.com/stackabletech/zookeeper-operator/pull/1106
 
 ## [26.7.0] - 2026-07-21

rust/operator-binary/src/listener_addresses.rs

@@ -30,10 +30,7 @@ type Result<T, E = Error> = std::result::Result<T, E>;
 ///
 /// An address is a hostname or IP address of a node, a cluster IP or an external load balancer,
 /// depending on the Service type behind the Listener.
-///
-/// The [`Default`] value is empty, which renders as an empty connection string, see
-/// [`build_discovery_configmap`](crate::discovery::build_discovery_configmap).
-#[derive(Clone, Debug, Default, PartialEq, Eq)]
+#[derive(Clone, Debug, PartialEq, Eq)]
 pub struct ListenerAddresses(BTreeSet<(String, u16)>);
 
 impl ListenerAddresses {
@@ -50,8 +47,9 @@ impl ListenerAddresses {
 
 /// Reads the addresses that `listener` publishes for `port_name`.
 ///
-/// Returns `Ok(None)` while the Listener carries no ingress addresses at all, which is the normal
-/// state between creating the Listener and the listener operator publishing its addresses. A
+/// Returns `Ok(None)` while the Listener carries no ingress addresses at all
+/// (no status yet, or an empty address list), which is the normal state between
+/// creating the Listener and the listener operator publishing its addresses. A
 /// Listener that does publish addresses, but none for `port_name`, is an error.
 // TODO (@NickLarsenNZ): Move this to stackable-operator, so it can be used as
 // listener.addresses_for_port(port_name)
@@ -67,6 +65,12 @@ pub fn listener_addresses(
         return Ok(None);
     };
 
+    // The listener operator publishes an empty list until it knows an address, e.g. before the
+    // Pods behind a NodePort Listener are scheduled. That is "not ready yet", not a missing port.
+    if ingress_addresses.is_empty() {
+        return Ok(None);
+    }
+
     let address_port_pairs = ingress_addresses
         .iter()
         // Filter the addresses that have the port we are interested in (they likely all have it though)
@@ -163,6 +167,15 @@ mod tests {
         );
     }
 
+    #[test]
+    fn listener_addresses_with_empty_ingress_list_is_not_ready_yet() {
+        assert_eq!(
+            listener_addresses(&role_listener(Some(vec![])), ZOOKEEPER_SERVER_PORT_NAME)
+                .expect("addresses"),
+            None
+        );
+    }
+
     #[test]
     fn listener_addresses_missing_port_name_is_error() {
         let listener = role_listener(Some(vec![ingress(2181)]));

rust/operator-binary/src/zk_controller.rs

@@ -115,8 +115,8 @@ pub struct KubernetesResources<T> {
     pub listeners: Vec<Listener>,
     pub config_maps: Vec<ConfigMap>,
     /// The discovery `ConfigMap`, kept apart from the role group `config_maps` because the cluster
-    /// status carries a hash of it.
-    pub discovery_config_map: ConfigMap,
+    /// status carries a hash of it. `None` while none can be built or re-emitted.
+    pub discovery_config_map: Option<ConfigMap>,
     pub pod_disruption_budgets: Vec<PodDisruptionBudget>,
     pub service_accounts: Vec<ServiceAccount>,
     pub role_bindings: Vec<RoleBinding>,
@@ -190,7 +190,7 @@ pub fn error_policy(
 pub(crate) mod test_support {
     use stackable_operator::{
         cli::OperatorEnvironmentOptions, commons::networking::DomainName, crd::listener,
-        utils::cluster_info::KubernetesClusterInfo,
+        k8s_openapi::api::core::v1::ConfigMap, utils::cluster_info::KubernetesClusterInfo,
     };
 
     use crate::{
@@ -201,6 +201,9 @@ pub(crate) mod test_support {
         },
     };
 
+    /// The UID that `minimal_zk` gives the fixture cluster.
+    pub const CLUSTER_UID: &str = "c27b3971-ca72-42c1-80a4-abdfc1db0ddd";
+
     /// The expected `app.kubernetes.io/version` label value for the given product version.
     ///
     /// The `-stackable` suffix carries the operator's own version, which is `0.0.0-dev` on main
@@ -225,7 +228,7 @@ pub(crate) mod test_support {
             .get_or_insert_with(|| "default".to_owned());
         zk.metadata
             .uid
-            .get_or_insert_with(|| "c27b3971-ca72-42c1-80a4-abdfc1db0ddd".to_owned());
+            .get_or_insert_with(|| CLUSTER_UID.to_owned());
         zk
     }
 
@@ -248,20 +251,22 @@ pub(crate) mod test_support {
     pub fn try_validate(
         zk: &v1alpha1::ZookeeperCluster,
     ) -> Result<ValidatedCluster, super::validate::Error> {
-        try_validate_with_role_listener(zk, None)
+        try_validate_with_role_listener(zk, None, None)
     }
 
     /// Runs the real validate step against a minimal (auth-free) fixture and the given role
     /// Listener, returning the result so tests can assert on validation errors.
     pub fn try_validate_with_role_listener(
         zk: &v1alpha1::ZookeeperCluster,
         maybe_role_listener: Option<listener::v1alpha1::Listener>,
+        existing_discovery_config_map: Option<ConfigMap>,
     ) -> Result<ValidatedCluster, super::validate::Error> {
         validate(
             zk,
             &DereferencedObjects {
                 authentication_classes: DereferencedAuthenticationClasses::new_for_tests(),
                 maybe_role_listener,
+                existing_discovery_config_map,
             },
             &operator_environment(),
         )

rust/operator-binary/src/zk_controller/apply.rs

@@ -95,7 +95,10 @@ impl<'a> Applier<'a> {
         let services = self.add_resources(services).await?;
         let listeners = self.add_resources(listeners).await?;
         let config_maps = self.add_resources(config_maps).await?;
-        let discovery_config_map = self.add_resource(discovery_config_map).await?;
+        let discovery_config_map = match discovery_config_map {
+            Some(config_map) => Some(self.add_resource(config_map).await?),
+            None => None,
+        };
         let pod_disruption_budgets = self.add_resources(pod_disruption_budgets).await?;
         let stateful_sets = self.add_resources(stateful_sets).await?;
 

rust/operator-binary/src/zk_controller/build.rs

@@ -81,8 +81,8 @@ pub enum Error {
 /// [`Listener`](stackable_operator::crd::listener::v1alpha1::Listener) that were dereferenced and
 /// validated into
 /// [`ValidatedCluster::discovery_addresses`](ValidatedCluster#structfield.discovery_addresses); see
-/// [`build_discovery_configmap`](discovery::build_discovery_configmap) for how its content depends
-/// on them.
+/// [`build_discovery_configmap`](discovery::build_discovery_configmap) for when it is built fresh,
+/// re-emitted or skipped.
 pub fn build(
     cluster: &ValidatedCluster,
     cluster_info: &KubernetesClusterInfo,
@@ -139,8 +139,7 @@ pub fn build(
     let listeners = vec![build_role_listener(cluster, &zk_role)];
 
     let discovery_config_map =
-        discovery::build_discovery_configmap(cluster, &zk_role, &cluster.discovery_addresses)
-            .context(DiscoveryConfigMapSnafu)?;
+        discovery::build_discovery_configmap(cluster, &zk_role).context(DiscoveryConfigMapSnafu)?;
 
     Ok(KubernetesResources {
         stateful_sets,
@@ -321,12 +320,9 @@ mod tests {
                 "simple-zookeeper-server-secondary",
             ]
         );
-        // The discovery ConfigMap is named after the cluster, and written even though the fixture
-        // has no role Listener publishing addresses yet (see `build()`).
-        assert_eq!(
-            resources.discovery_config_map.meta().name.as_deref(),
-            Some("simple-zookeeper")
-        );
+        // The fixture's role Listener does not exist and no discovery
+        // ConfigMap is stored, so none is built yet.
+        assert!(resources.discovery_config_map.is_none());
         // The single role-level Listener for the one ZooKeeper role (`server`).
         assert_eq!(
             sorted_names(&resources.listeners),

rust/operator-binary/src/zk_controller/build/resource/rbac.rs

@@ -33,7 +33,9 @@ mod tests {
     use serde_json::json;
 
     use super::*;
-    use crate::zk_controller::test_support::{app_version_label, minimal_zk, validated_cluster};
+    use crate::zk_controller::test_support::{
+        CLUSTER_UID, app_version_label, minimal_zk, validated_cluster,
+    };
 
     // `simple-zookeeper` vs `zookeeper`: see the swap-guard note on `minimal_zk`.
     fn cluster() -> ValidatedCluster {
@@ -81,7 +83,7 @@ mod tests {
                             "controller": true,
                             "kind": "ZookeeperCluster",
                             "name": "simple-zookeeper",
-                            "uid": "c27b3971-ca72-42c1-80a4-abdfc1db0ddd"
+                            "uid": CLUSTER_UID.to_owned()
                         }
                     ]
                 }
@@ -114,7 +116,7 @@ mod tests {
                             "controller": true,
                             "kind": "ZookeeperCluster",
                             "name": "simple-zookeeper",
-                            "uid": "c27b3971-ca72-42c1-80a4-abdfc1db0ddd"
+                            "uid": CLUSTER_UID.to_owned()
                         }
                     ]
                 },

rust/operator-binary/src/zk_controller/build/resource/service.rs

@@ -129,7 +129,9 @@ mod tests {
     use super::*;
     use crate::{
         crd::ZookeeperRole,
-        zk_controller::test_support::{app_version_label, minimal_zk, validated_cluster},
+        zk_controller::test_support::{
+            CLUSTER_UID, app_version_label, minimal_zk, validated_cluster,
+        },
     };
 
     /// Every metrics Service must carry the Prometheus scrape label and the
@@ -193,7 +195,7 @@ mod tests {
                             "controller": true,
                             "kind": "ZookeeperCluster",
                             "name": "simple-zookeeper",
-                            "uid": "c27b3971-ca72-42c1-80a4-abdfc1db0ddd"
+                            "uid": CLUSTER_UID.to_owned()
                         }
                     ]
                 },

rust/operator-binary/src/zk_controller/dereference.rs

@@ -9,16 +9,23 @@ use snafu::{ResultExt, Snafu};
 use stackable_operator::{
     client::Client,
     crd::listener,
+    k8s_openapi::api::core::v1::ConfigMap,
     v2::{
-        controller_utils::{get_cluster_name, get_namespace},
-        types::{kubernetes::NamespaceName, operator::ClusterName},
+        controller_utils::{get_cluster_name, get_namespace, get_uid},
+        types::{
+            kubernetes::{NamespaceName, Uid},
+            operator::ClusterName,
+        },
     },
 };
 
-use crate::crd::{
-    ZookeeperRole,
-    authentication::{self, DereferencedAuthenticationClasses},
-    role_listener_name, v1alpha1,
+use crate::{
+    crd::{
+        ZookeeperRole,
+        authentication::{self, DereferencedAuthenticationClasses},
+        role_listener_name, v1alpha1,
+    },
+    discovery::discovery_config_map_name,
 };
 
 #[derive(Snafu, Debug)]
@@ -40,6 +47,17 @@ pub enum Error {
     FetchRoleListener {
         source: stackable_operator::client::Error,
     },
+
+    #[snafu(display("failed to fetch the discovery ConfigMap {config_map}"))]
+    FetchExistingDiscoveryConfigMap {
+        source: stackable_operator::client::Error,
+        config_map: String,
+    },
+
+    #[snafu(display("failed to get the cluster UID"))]
+    GetUid {
+        source: stackable_operator::v2::controller_utils::Error,
+    },
 }
 
 type Result<T, E = Error> = std::result::Result<T, E>;
@@ -55,9 +73,15 @@ pub struct DereferencedObjects {
     /// this object, so it can only be built once the Listener exists and carries them. The
     /// controller watches Listeners, so a reconciliation is triggered as soon as that happens.
     pub maybe_role_listener: Option<listener::v1alpha1::Listener>,
+
+    /// The discovery ConfigMap as stored by an earlier reconciliation (named by
+    /// [`discovery_config_map_name`]), or `None` while none owned by this cluster is stored.
+    /// The build step re-emits it while the role Listener has no usable address.
+    pub existing_discovery_config_map: Option<ConfigMap>,
 }
 
-/// Fetches all Kubernetes objects referenced from the [`v1alpha1::ZookeeperCluster`] spec.
+/// Fetches all Kubernetes objects referenced from the [`v1alpha1::ZookeeperCluster`] spec
+/// (only considering a ConfigMap owned by this cluster).
 pub async fn dereference(
     client: &Client,
     zk: &v1alpha1::ZookeeperCluster,
@@ -73,10 +97,21 @@ pub async fn dereference(
     .context(FetchAuthenticationClassesSnafu)?;
 
     let maybe_role_listener = fetch_role_listener(client, &cluster_name, &namespace).await?;
+    let uid = get_uid(zk).context(GetUidSnafu)?;
+
+    let discovery_config_map_name = discovery_config_map_name(&cluster_name);
+    let existing_discovery_config_map = client
+        .get_opt::<ConfigMap>(discovery_config_map_name.as_ref(), namespace.as_ref())
+        .await
+        .context(FetchExistingDiscoveryConfigMapSnafu {
+            config_map: discovery_config_map_name.as_ref(),
+        })?
+        .filter(|config_map| is_owned_by(config_map, &uid));
 
     Ok(DereferencedObjects {
         authentication_classes,
         maybe_role_listener,
+        existing_discovery_config_map,
     })
 }
 
@@ -92,3 +127,72 @@ async fn fetch_role_listener(
         .await
         .context(FetchRoleListenerSnafu)
 }
+
+/// Whether `config_map` is owned by the object with the given UID.
+///
+/// A ConfigMap of the same name left behind by a deleted cluster (e.g. after
+/// `kubectl delete --cascade=orphan`) must not be re-emitted for a new cluster.
+fn is_owned_by(config_map: &ConfigMap, uid: &Uid) -> bool {
+    config_map
+        .metadata
+        .owner_references
+        .iter()
+        .flatten()
+        .any(|owner_reference| owner_reference.uid == uid.to_string())
+}
+
+#[cfg(test)]
+mod tests {
+    use std::str::FromStr;
+
+    use stackable_operator::k8s_openapi::apimachinery::pkg::apis::meta::v1::{
+        ObjectMeta, OwnerReference,
+    };
+
+    use super::*;
+    use crate::zk_controller::test_support::CLUSTER_UID;
+
+    fn config_map_owned_by(owner_uids: &[&str]) -> ConfigMap {
+        ConfigMap {
+            metadata: ObjectMeta {
+                owner_references: Some(
+                    owner_uids
+                        .iter()
+                        .map(|uid| OwnerReference {
+                            uid: (*uid).to_owned(),
+                            ..OwnerReference::default()
+                        })
+                        .collect(),
+                ),
+                ..ObjectMeta::default()
+            },
+            ..ConfigMap::default()
+        }
+    }
+
+    fn cluster_uid() -> Uid {
+        Uid::from_str(CLUSTER_UID).expect("valid UID")
+    }
+
+    #[test]
+    fn config_map_owned_by_this_cluster_is_considered() {
+        assert!(is_owned_by(
+            &config_map_owned_by(&[CLUSTER_UID]),
+            &cluster_uid()
+        ));
+    }
+
+    /// E.g. left behind by a deleted cluster of the same name.
+    #[test]
+    fn config_map_owned_by_another_object_is_ignored() {
+        assert!(!is_owned_by(
+            &config_map_owned_by(&["00000000-0000-0000-0000-000000000000"]),
+            &cluster_uid()
+        ));
+    }
+
+    #[test]
+    fn config_map_without_owner_references_is_ignored() {
+        assert!(!is_owned_by(&ConfigMap::default(), &cluster_uid()));
+    }
+}