fix: Add missing owns/watches and test early-exit reconcile

CHANGELOG.md

@@ -37,13 +37,16 @@
 - Fix a longstanding problem of including empty `categories`, `shortNames` and `additionalPrinterColumns` in the CRDs,
   which could cause problems with GitOps tools (e.g. ArgoCD) reporting a diff in the custom resources.
   See [our internal issue](https://github.com/stackabletech/hdfs-operator/issues/626) and [the fix](https://github.com/kube-rs/kube/pull/2042) for details ([#773]).
+- The operator now watches all resources that it creates and early-exits the reconcile action when the
+  cluster is marked for deletion ([#781]).
 
 [#756]: https://github.com/stackabletech/superset-operator/pull/756
 [#761]: https://github.com/stackabletech/superset-operator/pull/761
 [#765]: https://github.com/stackabletech/superset-operator/pull/765
 [#772]: https://github.com/stackabletech/superset-operator/pull/772
 [#773]: https://github.com/stackabletech/superset-operator/pull/773
 [#779]: https://github.com/stackabletech/superset-operator/pull/779
+[#781]: https://github.com/stackabletech/superset-operator/pull/781
 
 ## [26.7.0] - 2026-07-21
 

deploy/helm/superset-operator/templates/clusterrole-operator.yaml

@@ -13,15 +13,17 @@ rules:
       - nodes/proxy
     verbs:
       - get
-  # Manage core namespaced resources created per SupersetCluster.
-  # All resources are applied via Server-Side Apply (create + patch) and tracked for
-  # orphan cleanup (list + delete). ReconciliationPaused uses get.
-  # Secrets are needed for auto-generated SECRET_KEY (random_secret_creation).
+  # Manage core namespaced resources created per SupersetCluster (the ServiceAccount is
+  # also created per DruidConnection and provides workload pod identity).
+  # All resources are applied via Server-Side Apply (create + patch), tracked for
+  # orphan cleanup (list + delete) and watched by the controller. ReconciliationPaused
+  # uses get. Secrets are needed for auto-generated SECRET_KEY (random_secret_creation).
   - apiGroups:
       - ""
     resources:
       - configmaps
       - secrets
+      - serviceaccounts
       - services
     verbs:
       - create
@@ -30,20 +32,9 @@ rules:
       - list
       - patch
       - watch
-  # ServiceAccount created per SupersetCluster and per DruidConnection.
-  # Applied via SSA and tracked for orphan cleanup.
-  - apiGroups:
-      - ""
-    resources:
-      - serviceaccounts
-    verbs:
-      - create
-      - delete
-      - get
-      - list
-      - patch
   # RoleBinding created per SupersetCluster to bind the product ClusterRole to the workload
-  # ServiceAccount. Applied via SSA and tracked for orphan cleanup.
+  # ServiceAccount. Applied via SSA and tracked for orphan cleanup and watched by the
+  # controller.
   - apiGroups:
       - rbac.authorization.k8s.io
     resources:
@@ -54,6 +45,7 @@ rules:
       - get
       - list
       - patch
+      - watch
   # Required to bind the product ClusterRole to the per-cluster ServiceAccount.
   - apiGroups:
       - rbac.authorization.k8s.io
@@ -98,7 +90,8 @@ rules:
       - list
       - patch
       - watch
-  # PodDisruptionBudget created per role. Applied via SSA and tracked for orphan cleanup.
+  # PodDisruptionBudget created per role. Applied via SSA and tracked for orphan cleanup
+  # and watched by the controller.
   - apiGroups:
       - policy
     resources:
@@ -109,6 +102,7 @@ rules:
       - get
       - list
       - patch
+      - watch
   # Required for maintaining the CRDs within the operator (including the conversion webhook info).
   # Also for the startup condition check before the controller can run.
   - apiGroups:
@@ -167,7 +161,7 @@ rules:
       - list
       - watch
   # Listener created per role group for external access. Applied via SSA and tracked for orphan
-  # cleanup.
+  # cleanup and watched by the controller.
   - apiGroups:
       - listeners.stackable.tech
     resources:
@@ -178,3 +172,4 @@ rules:
       - get
       - list
       - patch
+      - watch

rust/operator-binary/src/controller.rs

@@ -363,6 +363,10 @@ pub async fn reconcile_superset(
 ) -> Result<Action> {
     tracing::info!("Starting reconcile");
 
+    if superset.meta().deletion_timestamp.is_some() {
+        return Ok(Action::await_change());
+    }
+
     let superset = superset
         .0
         .as_ref()
@@ -454,7 +458,16 @@ pub(crate) mod test_support {
 
 #[cfg(test)]
 mod controller_tests {
-    use super::{CONTROLLER_NAME, OPERATOR_NAME, PRODUCT_NAME};
+    use std::str::FromStr;
+
+    use stackable_operator::{
+        client::Client,
+        commons::networking::DomainName,
+        kube::{Client as KubeClient, Config},
+        utils::cluster_info::KubernetesClusterInfo,
+    };
+
+    use super::{CONTROLLER_NAME, OPERATOR_NAME, PRODUCT_NAME, *};
 
     #[test]
     fn test_constants() {
@@ -463,4 +476,54 @@ mod controller_tests {
         let _ = *OPERATOR_NAME;
         let _ = *CONTROLLER_NAME;
     }
+
+    /// The client points at a closed port, so any API call would fail the reconciliation: an `Ok`
+    /// proves that a cluster being deleted returns before the reconciler touches the Kubernetes
+    /// API, and because the spec is invalid, before the [`DeserializeGuard`] is unwrapped.
+    #[test]
+    fn reconcile_exits_early_for_deleted_cluster() {
+        let superset = serde_yaml::from_str(
+            r#"
+apiVersion: superset.stackable.tech/v1alpha1
+kind: SupersetCluster
+metadata:
+  name: superset
+  namespace: default
+  deletionTimestamp: "2026-08-14T12:00:00Z"
+spec: {}
+"#,
+        )
+        .expect("YAML parses; the invalid spec is captured inside the DeserializeGuard");
+
+        let action = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("current-thread tokio runtime")
+            .block_on(async {
+                let ctx = Arc::new(Ctx {
+                    client: Client::new(
+                        KubeClient::try_from(Config::new(
+                            "http://127.0.0.1:1".parse().expect("valid static URI"),
+                        ))
+                        .expect("client from static config"),
+                        None,
+                        "default".to_owned(),
+                        KubernetesClusterInfo {
+                            cluster_domain: DomainName::from_str("cluster.local")
+                                .expect("valid cluster domain"),
+                        },
+                    ),
+                    operator_environment: OperatorEnvironmentOptions {
+                        operator_namespace: "stackable-operators".to_owned(),
+                        operator_service_name: "superset-operator".to_owned(),
+                        image_repository: "oci.stackable.tech/sdp".to_owned(),
+                    },
+                });
+
+                reconcile_superset(Arc::new(superset), ctx).await
+            })
+            .expect("a deleted cluster reconciles without any API call");
+
+        assert_eq!(action, Action::await_change());
+    }
 }

rust/operator-binary/src/druid_connection_controller/mod.rs

@@ -17,7 +17,7 @@ use stackable_operator::{
         core::v1::{ConfigMap, EnvVar, EnvVarSource, PodSpec, PodTemplateSpec, SecretKeySelector},
     },
     kube::{
-        ResourceExt,
+        Resource, ResourceExt,
         core::{DeserializeGuard, DynamicObject, error_boundary},
         runtime::{controller::Action, reflector::ObjectRef},
     },
@@ -154,6 +154,10 @@ pub async fn reconcile_druid_connection(
 ) -> Result<Action> {
     tracing::info!("Starting reconciling DruidConnections");
 
+    if druid_connection.meta().deletion_timestamp.is_some() {
+        return Ok(Action::await_change());
+    }
+
     let druid_connection = druid_connection
         .0
         .as_ref()
@@ -457,10 +461,64 @@ pub fn error_policy(
 
 #[cfg(test)]
 mod tests {
-    use stackable_operator::utils::yaml_from_str_singleton_map;
+    use stackable_operator::{
+        commons::networking::DomainName,
+        kube::{Client as KubeClient, Config},
+        utils::{cluster_info::KubernetesClusterInfo, yaml_from_str_singleton_map},
+    };
 
     use super::*;
 
+    /// The client points at a closed port, so any API call would fail the reconciliation: an `Ok`
+    /// proves that a connection being deleted returns before the reconciler touches the Kubernetes
+    /// API, and because the spec is invalid, before the [`DeserializeGuard`] is unwrapped.
+    #[test]
+    fn reconcile_exits_early_for_deleted_connection() {
+        let druid_connection = serde_yaml::from_str(
+            r#"
+apiVersion: superset.stackable.tech/v1alpha1
+kind: DruidConnection
+metadata:
+  name: simple-connection
+  namespace: default
+  deletionTimestamp: "2026-08-14T12:00:00Z"
+spec: {}
+"#,
+        )
+        .expect("YAML parses; the invalid spec is captured inside the DeserializeGuard");
+
+        let action = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("current-thread tokio runtime")
+            .block_on(async {
+                let ctx = Arc::new(Ctx {
+                    client: Client::new(
+                        KubeClient::try_from(Config::new(
+                            "http://127.0.0.1:1".parse().expect("valid static URI"),
+                        ))
+                        .expect("client from static config"),
+                        None,
+                        "default".to_owned(),
+                        KubernetesClusterInfo {
+                            cluster_domain: DomainName::from_str("cluster.local")
+                                .expect("valid cluster domain"),
+                        },
+                    ),
+                    operator_environment: OperatorEnvironmentOptions {
+                        operator_namespace: "stackable-operators".to_owned(),
+                        operator_service_name: "superset-operator".to_owned(),
+                        image_repository: "oci.stackable.tech/sdp".to_owned(),
+                    },
+                });
+
+                reconcile_druid_connection(Arc::new(druid_connection), ctx).await
+            })
+            .expect("a deleted connection reconciles without any API call");
+
+        assert_eq!(action, Action::await_change());
+    }
+
     #[test]
     fn test_constants() {
         // Test that dereferencing the constants does not panic.

rust/operator-binary/src/main.rs

@@ -9,12 +9,14 @@ use futures::{FutureExt, StreamExt, TryFutureExt};
 use stackable_operator::{
     YamlSchema,
     cli::{Command, RunArguments},
-    crd::authentication::core,
+    crd::{authentication::core, listener},
     eos::EndOfSupportChecker,
     k8s_openapi::api::{
         apps::v1::{Deployment, StatefulSet},
         batch::v1::Job,
-        core::v1::{ConfigMap, Service},
+        core::v1::{ConfigMap, Service, ServiceAccount},
+        policy::v1::PodDisruptionBudget,
+        rbac::v1::RoleBinding,
     },
     kube::{
         CustomResourceExt as _, ResourceExt,
@@ -134,17 +136,38 @@ async fn main() -> anyhow::Result<()> {
             let authentication_class_store = superset_controller.store();
             let config_map_store = superset_controller.store();
             let superset_controller = superset_controller
+                .owns(
+                    watch_namespace.get_api::<DeserializeGuard<ConfigMap>>(&client),
+                    watcher::Config::default(),
+                )
+                // Required for workers and beat.
+                .owns(
+                    watch_namespace.get_api::<DeserializeGuard<Deployment>>(&client),
+                    watcher::Config::default(),
+                )
+                .owns(
+                    watch_namespace
+                        .get_api::<DeserializeGuard<listener::v1alpha1::Listener>>(&client),
+                    watcher::Config::default(),
+                )
+                .owns(
+                    watch_namespace.get_api::<DeserializeGuard<PodDisruptionBudget>>(&client),
+                    watcher::Config::default(),
+                )
+                .owns(
+                    watch_namespace.get_api::<DeserializeGuard<RoleBinding>>(&client),
+                    watcher::Config::default(),
+                )
                 .owns(
                     watch_namespace.get_api::<DeserializeGuard<Service>>(&client),
                     watcher::Config::default(),
                 )
                 .owns(
-                    watch_namespace.get_api::<DeserializeGuard<StatefulSet>>(&client),
+                    watch_namespace.get_api::<DeserializeGuard<ServiceAccount>>(&client),
                     watcher::Config::default(),
                 )
-                // Required for workers and beat.
                 .owns(
-                    watch_namespace.get_api::<DeserializeGuard<Deployment>>(&client),
+                    watch_namespace.get_api::<DeserializeGuard<StatefulSet>>(&client),
                     watcher::Config::default(),
                 )
                 .watches(