Merge pull request #33 from osodevops/fix/issue-30-orphaned-pods

sionsmith · web-flow · commit a55773055f68 · 2026-06-12T08:19:04.000+01:00
fix: Background-propagation cleanup (#30) + configurable backoffLimit (#31)
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -2,6 +2,20 @@
 
 All notable changes to this project will be documented in this file.
 
+## 0.2.10 - 2026-06-12
+
+### Added
+
+- Add `spec.backoffLimit` to `KafkaRestore` and `KafkaBackup` to control pod retries on generated Jobs (including scheduled CronJob runs). Fixes [#31](https://github.com/osodevops/strimzi-backup-operator/issues/31).
+
+### Changed
+
+- Restore Jobs now default to `backoffLimit: 0` (a single attempt, previously 3): restores append to or purge target topics, so retrying a partially completed attempt can duplicate data. Set `spec.backoffLimit` explicitly to opt into retries. Backup Jobs keep the previous default of 3.
+
+### Fixed
+
+- Delete Jobs, CronJobs, and ConfigMaps with explicit Background propagation when a `KafkaBackup`/`KafkaRestore` is deleted. The batch/v1 Job API's legacy default deletion propagation is `Orphan`, which stripped the Job ownerReference from its pods and left Completed pods behind. Fixes [#30](https://github.com/osodevops/strimzi-backup-operator/issues/30).
+
 ## 0.2.9 - 2026-06-12
 
 ### Fixed
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "kafka-backup-operator"
-version = "0.2.9"
+version = "0.2.10"
 edition = "2021"
 rust-version = "1.88"
 license = "Apache-2.0"
diff --git a/deploy/crds/kafkabackups.yaml b/deploy/crds/kafkabackups.yaml
@@ -96,6 +96,12 @@ spec:
                 required:
                 - type
                 type: object
+              backoffLimit:
+                description: Number of pod retries before a backup Job is marked failed (`spec.backoffLimit` on generated Jobs, including scheduled CronJob runs). Defaults to 3.
+                format: int32
+                minimum: 0.0
+                nullable: true
+                type: integer
               backup:
                 description: Backup options (compression, encryption, parallelism)
                 nullable: true
diff --git a/deploy/crds/kafkarestores.yaml b/deploy/crds/kafkarestores.yaml
@@ -93,6 +93,12 @@ spec:
                 required:
                 - type
                 type: object
+              backoffLimit:
+                description: 'Number of pod retries before the restore Job is marked failed (`spec.backoffLimit` on the generated Job). Defaults to 0 so a restore runs exactly once: restores append to or purge target topics, so a retry of a partially completed attempt can duplicate data and must be opted into deliberately.'
+                format: int32
+                minimum: 0.0
+                nullable: true
+                type: integer
               backupRef:
                 description: Reference to the source backup
                 properties:
diff --git a/deploy/helm/strimzi-backup-operator/Chart.yaml b/deploy/helm/strimzi-backup-operator/Chart.yaml
@@ -2,8 +2,8 @@ apiVersion: v2
 name: strimzi-backup-operator
 description: Kubernetes operator for Kafka backup and restore, integrated with Strimzi
 type: application
-version: 0.2.9
-appVersion: "0.2.9"
+version: 0.2.10
+appVersion: "0.2.10"
 home: https://github.com/osodevops/strimzi-backup-operator
 sources:
   - https://github.com/osodevops/strimzi-backup-operator
diff --git a/deploy/helm/strimzi-backup-operator/crds/kafkabackups.yaml b/deploy/helm/strimzi-backup-operator/crds/kafkabackups.yaml
@@ -96,6 +96,12 @@ spec:
                 required:
                 - type
                 type: object
+              backoffLimit:
+                description: Number of pod retries before a backup Job is marked failed (`spec.backoffLimit` on generated Jobs, including scheduled CronJob runs). Defaults to 3.
+                format: int32
+                minimum: 0.0
+                nullable: true
+                type: integer
               backup:
                 description: Backup options (compression, encryption, parallelism)
                 nullable: true
diff --git a/deploy/helm/strimzi-backup-operator/crds/kafkarestores.yaml b/deploy/helm/strimzi-backup-operator/crds/kafkarestores.yaml
@@ -93,6 +93,12 @@ spec:
                 required:
                 - type
                 type: object
+              backoffLimit:
+                description: 'Number of pod retries before the restore Job is marked failed (`spec.backoffLimit` on the generated Job). Defaults to 0 so a restore runs exactly once: restores append to or purge target topics, so a retry of a partially completed attempt can duplicate data and must be opted into deliberately.'
+                format: int32
+                minimum: 0.0
+                nullable: true
+                type: integer
               backupRef:
                 description: Reference to the source backup
                 properties:
diff --git a/src/adapters/backup_config.rs b/src/adapters/backup_config.rs
@@ -507,6 +507,7 @@ mod tests {
             resources: None,
             template: None,
             image: None,
+            backoff_limit: None,
             consumer_groups: None,
         };
         let mut backup = KafkaBackup::new("test-backup", spec);
diff --git a/src/crd/kafka_backup.rs b/src/crd/kafka_backup.rs
@@ -89,6 +89,13 @@ pub struct KafkaBackupSpec {
     /// Container image for the backup job (default: osodevops/kafka-backup:v0.15.6)
     #[serde(skip_serializing_if = "Option::is_none")]
     pub image: Option<String>,
+
+    /// Number of pod retries before a backup Job is marked failed
+    /// (`spec.backoffLimit` on generated Jobs, including scheduled CronJob
+    /// runs). Defaults to 3.
+    #[schemars(range(min = 0))]
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub backoff_limit: Option<i32>,
 }
 
 /// Backup-specific options
diff --git a/src/crd/kafka_restore.rs b/src/crd/kafka_restore.rs
@@ -85,6 +85,15 @@ pub struct KafkaRestoreSpec {
     /// Container image for the restore job (default: osodevops/kafka-backup:v0.15.6)
     #[serde(skip_serializing_if = "Option::is_none")]
     pub image: Option<String>,
+
+    /// Number of pod retries before the restore Job is marked failed
+    /// (`spec.backoffLimit` on the generated Job). Defaults to 0 so a restore
+    /// runs exactly once: restores append to or purge target topics, so a
+    /// retry of a partially completed attempt can duplicate data and must be
+    /// opted into deliberately.
+    #[schemars(range(min = 0))]
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub backoff_limit: Option<i32>,
 }
 
 /// Reference to a KafkaBackup CR and optional specific backup snapshot
diff --git a/src/jobs/backup_job.rs b/src/jobs/backup_job.rs
@@ -99,7 +99,7 @@ pub fn build_backup_job(
             ..Default::default()
         },
         spec: Some(JobSpec {
-            backoff_limit: Some(3),
+            backoff_limit: Some(backup.spec.backoff_limit.unwrap_or(3)),
             template: PodTemplateSpec {
                 metadata: Some(ObjectMeta {
                     labels: Some(build_labels(&cr_name, &cluster.name, "backup")),
@@ -163,6 +163,7 @@ mod tests {
             resources: None,
             template: None,
             image: None,
+            backoff_limit: None,
         };
 
         let mut backup = KafkaBackup::new("test-backup", spec);
diff --git a/src/jobs/cronjob.rs b/src/jobs/cronjob.rs
@@ -108,7 +108,7 @@ pub fn build_backup_cronjob(
                     ..Default::default()
                 }),
                 spec: Some(JobSpec {
-                    backoff_limit: Some(3),
+                    backoff_limit: Some(backup.spec.backoff_limit.unwrap_or(3)),
                     template: PodTemplateSpec {
                         metadata: Some(ObjectMeta {
                             labels: Some(labels),
diff --git a/src/jobs/restore_job.rs b/src/jobs/restore_job.rs
@@ -103,7 +103,9 @@ pub fn build_restore_job(
             ..Default::default()
         },
         spec: Some(JobSpec {
-            backoff_limit: Some(3),
+            // Restores append/purge data: default to a single attempt so a
+            // partially completed restore is never re-applied implicitly.
+            backoff_limit: Some(restore.spec.backoff_limit.unwrap_or(0)),
             template: PodTemplateSpec {
                 metadata: Some(ObjectMeta {
                     labels: Some(build_labels(&cr_name, &cluster.name, "restore")),
diff --git a/src/reconcilers/backup.rs b/src/reconcilers/backup.rs
@@ -19,7 +19,8 @@ use crate::jobs::cronjob::build_backup_cronjob;
 use crate::jobs::job_state::{classify_jobs, job_failed, job_succeeded, should_create_backup_job};
 use crate::metrics::prometheus::MetricsState;
 use crate::reconcilers::{
-    job_service_account_name, FINALIZER, TRIGGER_ANNOTATION, TRIGGER_VALUE_NOW,
+    cleanup_delete_params, job_service_account_name, FINALIZER, TRIGGER_ANNOTATION,
+    TRIGGER_VALUE_NOW,
 };
 use crate::retention::policy::evaluate_retention;
 use crate::retention::storage::{discover_backup_history, prune_backup_ids};
@@ -189,9 +190,7 @@ async fn handle_cleanup(backup: &KafkaBackup, client: &Client, namespace: &str)
     if let Ok(job_list) = jobs_api.list(&lp).await {
         for job in job_list {
             let job_name = job.metadata.name.unwrap_or_default();
-            let _ = jobs_api
-                .delete(&job_name, &kube::api::DeleteParams::default())
-                .await;
+            let _ = jobs_api.delete(&job_name, &cleanup_delete_params()).await;
         }
     }
 
@@ -200,15 +199,13 @@ async fn handle_cleanup(backup: &KafkaBackup, client: &Client, namespace: &str)
         Api::namespaced(client.clone(), namespace);
     let cronjob_name = format!("{name}-scheduled");
     let _ = cronjob_api
-        .delete(&cronjob_name, &kube::api::DeleteParams::default())
+        .delete(&cronjob_name, &cleanup_delete_params())
         .await;
 
     // Delete ConfigMap
     let cm_api: Api<ConfigMap> = Api::namespaced(client.clone(), namespace);
     let cm_name = format!("{name}-config");
-    let _ = cm_api
-        .delete(&cm_name, &kube::api::DeleteParams::default())
-        .await;
+    let _ = cm_api.delete(&cm_name, &cleanup_delete_params()).await;
 
     // Remove finalizer
     let backup_api: Api<KafkaBackup> = Api::namespaced(client.clone(), namespace);
diff --git a/src/reconcilers/mod.rs b/src/reconcilers/mod.rs
@@ -22,6 +22,16 @@ pub const JOB_SERVICE_ACCOUNT_ENV: &str = "BACKUP_JOB_SERVICE_ACCOUNT";
 /// `serviceAccount.name` overrides.
 pub const DEFAULT_JOB_SERVICE_ACCOUNT: &str = "strimzi-backup-operator";
 
+/// Delete parameters for cleaning up resources owned by a backup/restore CR
+/// (Jobs, CronJobs, ConfigMaps) when the CR is deleted.
+///
+/// Background propagation must be explicit: the batch/v1 Job API's legacy
+/// server-side default is `Orphan`, which strips the Job ownerReference from
+/// its pods instead of deleting them, leaving Completed pods behind.
+pub fn cleanup_delete_params() -> kube::api::DeleteParams {
+    kube::api::DeleteParams::background()
+}
+
 pub fn job_service_account_name() -> Option<String> {
     let value = std::env::var(JOB_SERVICE_ACCOUNT_ENV)
         .unwrap_or_else(|_| DEFAULT_JOB_SERVICE_ACCOUNT.to_string());
diff --git a/src/reconcilers/restore.rs b/src/reconcilers/restore.rs
@@ -16,7 +16,7 @@ use crate::error::{Error, Result};
 use crate::jobs::job_state::{classify_jobs, JobsState};
 use crate::jobs::restore_job::build_restore_job;
 use crate::metrics::prometheus::MetricsState;
-use crate::reconcilers::{job_service_account_name, FINALIZER};
+use crate::reconcilers::{cleanup_delete_params, job_service_account_name, FINALIZER};
 use crate::status::conditions::*;
 use crate::strimzi::kafka_cr::resolve_kafka_cluster;
 use crate::strimzi::kafka_user::resolve_auth;
@@ -232,18 +232,14 @@ async fn handle_cleanup(restore: &KafkaRestore, client: &Client, namespace: &str
     if let Ok(job_list) = jobs_api.list(&lp).await {
         for job in job_list {
             let job_name = job.metadata.name.unwrap_or_default();
-            let _ = jobs_api
-                .delete(&job_name, &kube::api::DeleteParams::default())
-                .await;
+            let _ = jobs_api.delete(&job_name, &cleanup_delete_params()).await;
         }
     }
 
     // Delete ConfigMap
     let cm_api: Api<ConfigMap> = Api::namespaced(client.clone(), namespace);
     let cm_name = format!("{name}-config");
-    let _ = cm_api
-        .delete(&cm_name, &kube::api::DeleteParams::default())
-        .await;
+    let _ = cm_api.delete(&cm_name, &cleanup_delete_params()).await;
 
     // Remove finalizer
     let restore_api: Api<KafkaRestore> = Api::namespaced(client.clone(), namespace);
diff --git a/tests/integration/backup_test.rs b/tests/integration/backup_test.rs
@@ -78,6 +78,7 @@ fn sample_backup() -> KafkaBackup {
         resources: None,
         template: None,
         image: None,
+        backoff_limit: None,
     };
     let mut backup = KafkaBackup::new("daily-backup", spec);
     backup.metadata.namespace = Some("kafka".to_string());
@@ -464,3 +465,61 @@ fn test_backup_job_falls_back_to_operator_service_account() {
         Some("strimzi-backup-operator")
     );
 }
+
+/// Backups are safe to retry; the historical default of 3 is kept.
+#[test]
+fn test_backup_job_default_backoff_limit_is_three() {
+    let backup = sample_backup();
+    let job = build_backup_job(
+        &backup,
+        "daily-backup-20260213-020000",
+        "daily-backup-config",
+        &sample_cluster(),
+        &ResolvedAuth::None,
+        None,
+    )
+    .unwrap();
+
+    assert_eq!(job.spec.as_ref().unwrap().backoff_limit, Some(3));
+}
+
+#[test]
+fn test_backup_job_custom_backoff_limit() {
+    let mut backup = sample_backup();
+    backup.spec.backoff_limit = Some(1);
+    let job = build_backup_job(
+        &backup,
+        "daily-backup-20260213-020000",
+        "daily-backup-config",
+        &sample_cluster(),
+        &ResolvedAuth::None,
+        None,
+    )
+    .unwrap();
+
+    assert_eq!(job.spec.as_ref().unwrap().backoff_limit, Some(1));
+}
+
+#[test]
+fn test_backup_cronjob_custom_backoff_limit() {
+    let mut backup = sample_backup();
+    backup.spec.backoff_limit = Some(0);
+    let cronjob = build_backup_cronjob(
+        &backup,
+        "daily-backup-config",
+        &sample_cluster(),
+        &ResolvedAuth::None,
+        None,
+    )
+    .unwrap();
+
+    let job_spec = cronjob
+        .spec
+        .as_ref()
+        .unwrap()
+        .job_template
+        .spec
+        .as_ref()
+        .unwrap();
+    assert_eq!(job_spec.backoff_limit, Some(0));
+}
diff --git a/tests/integration/cleanup_test.rs b/tests/integration/cleanup_test.rs
@@ -0,0 +1,22 @@
+use kube::api::PropagationPolicy;
+
+use kafka_backup_operator::reconcilers::cleanup_delete_params;
+
+/// Issue #30: deleting a KafkaRestore left its Job's pods behind. The batch/v1
+/// Job API's legacy server-side default deletion propagation is `Orphan`, so
+/// deleting a Job without an explicit propagation policy strips the Job
+/// ownerReference from its pods instead of deleting them (kubectl masks this
+/// by always sending Background). Cleanup deletes must request Background
+/// propagation so dependents are garbage collected.
+#[test]
+fn test_cleanup_deletes_propagate_to_dependents() {
+    let params = cleanup_delete_params();
+    assert!(
+        matches!(
+            params.propagation_policy,
+            Some(PropagationPolicy::Background)
+        ),
+        "cleanup must delete with Background propagation, got {:?}",
+        params.propagation_policy
+    );
+}
diff --git a/tests/integration/mod.rs b/tests/integration/mod.rs
@@ -1,4 +1,5 @@
 mod backup_test;
+mod cleanup_test;
 mod job_state_test;
 mod reconcile_backup_test;
 mod restore_test;
diff --git a/tests/integration/reconcile_backup_test.rs b/tests/integration/reconcile_backup_test.rs
@@ -53,6 +53,7 @@ fn scheduled_backup(suspend: bool) -> KafkaBackup {
         resources: None,
         template: None,
         image: None,
+        backoff_limit: None,
     };
     let mut backup = KafkaBackup::new("daily-backup", spec);
     backup.metadata.namespace = Some("kafka".to_string());
diff --git a/tests/integration/restore_test.rs b/tests/integration/restore_test.rs
