fix: suspending KafkaBackup now suspends the underlying CronJob

CHANGELOG.md

@@ -2,6 +2,12 @@
 
 All notable changes to this project will be documented in this file.
 
+## 0.2.7 - 2026-06-10
+
+### Fixed
+
+- Propagate `spec.schedule.suspend` to the scheduled backup CronJob so suspending a `KafkaBackup` stops scheduled runs, and report a `BackupSuspended` condition while suspended. Fixes [#24](https://github.com/osodevops/strimzi-backup-operator/issues/24).
+
 ## 0.2.6 - 2026-06-03
 
 ### Added

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "kafka-backup-operator"
-version = "0.2.6"
+version = "0.2.7"
 edition = "2021"
 rust-version = "1.88"
 license = "Apache-2.0"
@@ -56,3 +56,6 @@ regex = "1"
 tokio-test = "0.4"
 assert-json-diff = "2"
 tempfile = "3"
+tower-test = "0.4"
+http = "1"
+http-body-util = "0.1"

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.6
-appVersion: "0.2.6"
+version: 0.2.7
+appVersion: "0.2.7"
 home: https://github.com/osodevops/strimzi-backup-operator
 sources:
   - https://github.com/osodevops/strimzi-backup-operator

src/reconcilers/backup.rs

@@ -96,25 +96,29 @@ pub async fn reconcile_backup(
     // Step 5: Check for scheduled vs one-shot
     let job_service_account = job_service_account_name();
     if let Some(schedule) = &backup.spec.schedule {
-        if !schedule.suspend {
-            // Create CronJob
-            let cronjob = build_backup_cronjob(
-                &backup,
-                &config_map_name,
-                &kafka_cluster,
-                &resolved_auth,
-                job_service_account.as_deref(),
-            )?;
-            let cronjob_api: Api<k8s_openapi::api::batch::v1::CronJob> =
-                Api::namespaced(client.clone(), &namespace);
-            let cronjob_name = format!("{name}-scheduled");
-
-            apply_resource(&cronjob_api, &cronjob_name, &cronjob).await?;
-
-            // Update status
+        // Apply the CronJob even when suspended so the suspend flag reaches
+        // the live resource; skipping here would leave an existing CronJob
+        // running on its old schedule.
+        let cronjob = build_backup_cronjob(
+            &backup,
+            &config_map_name,
+            &kafka_cluster,
+            &resolved_auth,
+            job_service_account.as_deref(),
+        )?;
+        let cronjob_api: Api<k8s_openapi::api::batch::v1::CronJob> =
+            Api::namespaced(client.clone(), &namespace);
+        let cronjob_name = format!("{name}-scheduled");
+
+        apply_resource(&cronjob_api, &cronjob_name, &cronjob).await?;
+
+        // Update status
+        if schedule.suspend {
+            update_status_suspended(&backup_api, &name, generation).await?;
+            info!(%name, "CronJob suspended for scheduled backup");
+        } else {
             let next_backup = schedule.cron.clone();
             update_status_scheduled(&backup_api, &name, generation, &next_backup).await?;
-
             info!(%name, cron = %schedule.cron, "CronJob created/updated for scheduled backup");
         }
     }
@@ -467,6 +471,30 @@ async fn update_status_scheduled(
     patch_status(api, name, &status).await
 }
 
+async fn update_status_suspended(
+    api: &Api<KafkaBackup>,
+    name: &str,
+    generation: i64,
+) -> Result<()> {
+    // Manual merge patch: `nextScheduledBackup` must be an explicit null to be
+    // cleared, but `KafkaBackupStatus` skips `None` fields when serializing.
+    let condition = ready(REASON_BACKUP_SUSPENDED, "Backup schedule is suspended");
+    let patch = serde_json::json!({
+        "status": {
+            "conditions": [condition],
+            "observedGeneration": generation,
+            "nextScheduledBackup": null
+        }
+    });
+    api.patch_status(
+        name,
+        &PatchParams::apply("kafka-backup-operator"),
+        &Patch::Merge(&patch),
+    )
+    .await?;
+    Ok(())
+}
+
 async fn update_status_completed(
     api: &Api<KafkaBackup>,
     name: &str,

src/status/conditions.rs

@@ -20,6 +20,7 @@ pub const REASON_BACKUP_RUNNING: &str = "BackupRunning";
 pub const REASON_BACKUP_COMPLETED: &str = "BackupCompleted";
 pub const REASON_BACKUP_FAILED: &str = "BackupFailed";
 pub const REASON_BACKUP_SCHEDULED: &str = "BackupScheduled";
+pub const REASON_BACKUP_SUSPENDED: &str = "BackupSuspended";
 pub const REASON_RESTORE_RUNNING: &str = "RestoreRunning";
 pub const REASON_RESTORE_COMPLETED: &str = "RestoreCompleted";
 pub const REASON_RESTORE_FAILED: &str = "RestoreFailed";

tests/integration/backup_test.rs

@@ -262,6 +262,33 @@ fn test_backup_cronjob_uses_configured_service_account() {
             && env.value.as_deref() == Some("kafka_backup=debug,rdkafka=info")));
 }
 
+#[test]
+fn test_backup_cronjob_propagates_suspend() {
+    let mut backup = sample_backup();
+    let cluster = sample_cluster();
+
+    let cronjob = build_backup_cronjob(
+        &backup,
+        "daily-backup-config",
+        &cluster,
+        &ResolvedAuth::None,
+        None,
+    )
+    .unwrap();
+    assert_eq!(cronjob.spec.as_ref().unwrap().suspend, Some(false));
+
+    backup.spec.schedule.as_mut().unwrap().suspend = true;
+    let cronjob = build_backup_cronjob(
+        &backup,
+        "daily-backup-config",
+        &cluster,
+        &ResolvedAuth::None,
+        None,
+    )
+    .unwrap();
+    assert_eq!(cronjob.spec.as_ref().unwrap().suspend, Some(true));
+}
+
 #[test]
 fn test_backup_jobs_apply_template_host_aliases() {
     let mut backup = sample_backup();

tests/integration/mod.rs

@@ -1,2 +1,3 @@
 mod backup_test;
+mod reconcile_backup_test;
 mod restore_test;

tests/integration/reconcile_backup_test.rs

@@ -0,0 +1,189 @@
+use std::sync::{Arc, Mutex};
+
+use http::{Request, Response};
+use http_body_util::BodyExt;
+use kafka_backup_operator::crd::common::*;
+use kafka_backup_operator::crd::kafka_backup::*;
+use kafka_backup_operator::crd::KafkaBackup;
+use kafka_backup_operator::metrics::prometheus::MetricsState;
+use kafka_backup_operator::reconcilers::backup::reconcile_backup;
+use kube::client::Body;
+use kube::Client;
+use serde_json::json;
+use tower_test::mock;
+
+#[derive(Debug)]
+struct RecordedRequest {
+    method: String,
+    path: String,
+    body: serde_json::Value,
+}
+
+fn scheduled_backup(suspend: bool) -> KafkaBackup {
+    let spec = KafkaBackupSpec {
+        strimzi_cluster_ref: StrimziClusterRef {
+            name: "production-cluster".to_string(),
+            namespace: None,
+            ca_secret: None,
+        },
+        authentication: None,
+        topics: None,
+        connection: None,
+        consumer_groups: None,
+        logging: None,
+        env: Vec::new(),
+        storage: StorageSpec {
+            storage_type: StorageType::Filesystem,
+            s3: None,
+            azure: None,
+            gcs: None,
+            filesystem: Some(FilesystemStorageSpec {
+                path: "/backups".to_string(),
+            }),
+        },
+        backup: None,
+        metrics: None,
+        offset_storage: None,
+        schedule: Some(ScheduleSpec {
+            cron: "0 2 * * *".to_string(),
+            timezone: None,
+            suspend,
+        }),
+        retention: None,
+        resources: None,
+        template: None,
+        image: None,
+    };
+    let mut backup = KafkaBackup::new("daily-backup", spec);
+    backup.metadata.namespace = Some("kafka".to_string());
+    backup.metadata.uid = Some("test-uid-12345".to_string());
+    backup.metadata.generation = Some(2);
+    backup.metadata.finalizers = Some(vec!["kafkabackup.com/cleanup".to_string()]);
+    backup
+}
+
+/// Run `reconcile_backup` against a mock API server and record every request
+/// the operator makes.
+async fn reconcile_with_mock_api(backup: KafkaBackup) -> Vec<RecordedRequest> {
+    let (mock_service, mut handle) = mock::pair::<Request<Body>, Response<Body>>();
+    let recorded = Arc::new(Mutex::new(Vec::new()));
+
+    let backup_json = serde_json::to_value(&backup).unwrap();
+    let dispatcher = {
+        let recorded = Arc::clone(&recorded);
+        tokio::spawn(async move {
+            while let Some((request, send)) = handle.next_request().await {
+                let method = request.method().to_string();
+                let path = request.uri().path().to_string();
+                let bytes = request.into_body().collect().await.unwrap().to_bytes();
+                let body: serde_json::Value = if bytes.is_empty() {
+                    serde_json::Value::Null
+                } else {
+                    serde_json::from_slice(&bytes).unwrap()
+                };
+
+                let (status, response_body) = if path.ends_with("/kafkas/production-cluster") {
+                    (
+                        200,
+                        json!({
+                            "apiVersion": "kafka.strimzi.io/v1beta2",
+                            "kind": "Kafka",
+                            "metadata": {"name": "production-cluster", "namespace": "kafka"},
+                            "spec": {"kafka": {"replicas": 3, "listeners": [
+                                {"name": "plain", "port": 9092, "type": "internal", "tls": false}
+                            ]}}
+                        }),
+                    )
+                } else if path.contains("/secrets/") {
+                    (
+                        404,
+                        json!({
+                            "kind": "Status", "apiVersion": "v1", "metadata": {},
+                            "status": "Failure", "message": "secret not found",
+                            "reason": "NotFound", "code": 404
+                        }),
+                    )
+                } else if path.ends_with("/jobs") {
+                    (
+                        200,
+                        json!({"kind": "JobList", "apiVersion": "batch/v1", "metadata": {}, "items": []}),
+                    )
+                } else if path.contains("/kafkabackups/") {
+                    (200, backup_json.clone())
+                } else {
+                    // Server-side apply of ConfigMap/CronJob: echo the object back
+                    (200, body.clone())
+                };
+
+                recorded
+                    .lock()
+                    .unwrap()
+                    .push(RecordedRequest { method, path, body });
+
+                let response = Response::builder()
+                    .status(status)
+                    .header("content-type", "application/json")
+                    .body(Body::from(serde_json::to_vec(&response_body).unwrap()))
+                    .unwrap();
+                send.send_response(response);
+            }
+        })
+    };
+
+    let client = Client::new(mock_service, "kafka");
+    let metrics = MetricsState::new();
+    reconcile_backup(Arc::new(backup), client, &metrics)
+        .await
+        .expect("reconcile should succeed");
+
+    dispatcher.await.unwrap();
+    Arc::try_unwrap(recorded).unwrap().into_inner().unwrap()
+}
+
+fn find_cronjob_patch(requests: &[RecordedRequest]) -> &RecordedRequest {
+    requests
+        .iter()
+        .find(|r| r.method == "PATCH" && r.path.ends_with("/cronjobs/daily-backup-scheduled"))
+        .expect("reconcile must apply the CronJob whenever a schedule is set")
+}
+
+#[tokio::test]
+async fn test_reconcile_propagates_suspend_to_cronjob() {
+    let requests = reconcile_with_mock_api(scheduled_backup(true)).await;
+
+    // The CronJob must still be applied when suspended, otherwise a
+    // previously-created CronJob keeps firing on its old schedule.
+    let cronjob_patch = find_cronjob_patch(&requests);
+    assert_eq!(cronjob_patch.body["spec"]["suspend"], json!(true));
+
+    let status_patch = requests
+        .iter()
+        .find(|r| r.method == "PATCH" && r.path.ends_with("/kafkabackups/daily-backup/status"))
+        .expect("reconcile must update the KafkaBackup status");
+    assert_eq!(
+        status_patch.body["status"]["conditions"][0]["reason"],
+        json!("BackupSuspended")
+    );
+    let status = status_patch.body["status"].as_object().unwrap();
+    assert!(
+        status.contains_key("nextScheduledBackup") && status["nextScheduledBackup"].is_null(),
+        "suspending must clear nextScheduledBackup with an explicit null"
+    );
+}
+
+#[tokio::test]
+async fn test_reconcile_applies_active_cronjob_when_not_suspended() {
+    let requests = reconcile_with_mock_api(scheduled_backup(false)).await;
+
+    let cronjob_patch = find_cronjob_patch(&requests);
+    assert_eq!(cronjob_patch.body["spec"]["suspend"], json!(false));
+
+    let status_patch = requests
+        .iter()
+        .find(|r| r.method == "PATCH" && r.path.ends_with("/kafkabackups/daily-backup/status"))
+        .expect("reconcile must update the KafkaBackup status");
+    assert_eq!(
+        status_patch.body["status"]["conditions"][0]["reason"],
+        json!("BackupScheduled")
+    );
+}