Skip to content

Commit 017d23b

Browse files
authored
Merge pull request #32 from osodevops/fix/issue-29-restore-rerun
fix: stop re-running restore and one-shot backup jobs after completion
2 parents e4d13e4 + 905b401 commit 017d23b

12 files changed

Lines changed: 534 additions & 159 deletions

File tree

CHANGELOG.md

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,15 @@
22

33
All notable changes to this project will be documented in this file.
44

5+
## 0.2.9 - 2026-06-12
6+
7+
### Fixed
8+
9+
- Stop re-running `KafkaRestore` jobs after completion. A finished restore Job has `active=0`, which the reconciler read as "no job running" and re-created the Job on every 5-minute requeue; Job creation is now gated on the full set of Jobs for the resource (running, succeeded, or failed), and one-shot `KafkaBackup` runs are gated the same way. Fixes [#29](https://github.com/osodevops/strimzi-backup-operator/issues/29).
10+
- Watch backup/restore Jobs from the controllers so `KafkaBackup`/`KafkaRestore` status reflects Job completion or failure within seconds instead of after the next periodic requeue.
11+
- Treat a terminally failed restore Job (backoffLimit exhausted) as terminal: report a `RestoreFailed` condition instead of silently re-creating the Job every requeue. Pod-level retries remain owned by the Job's `backoffLimit`.
12+
- Make status patches idempotent so repeated reconciles no longer rewrite `lastTransitionTime`/`completionTime` with the current wall clock on every pass.
13+
514
## 0.2.8 - 2026-06-11
615

716
### Added

Cargo.lock

Lines changed: 1 addition & 1 deletion
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

Cargo.toml

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
[package]
22
name = "kafka-backup-operator"
3-
version = "0.2.8"
3+
version = "0.2.9"
44
edition = "2021"
55
rust-version = "1.88"
66
license = "Apache-2.0"

deploy/helm/strimzi-backup-operator/Chart.yaml

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2,8 +2,8 @@ apiVersion: v2
22
name: strimzi-backup-operator
33
description: Kubernetes operator for Kafka backup and restore, integrated with Strimzi
44
type: application
5-
version: 0.2.8
6-
appVersion: "0.2.8"
5+
version: 0.2.9
6+
appVersion: "0.2.9"
77
home: https://github.com/osodevops/strimzi-backup-operator
88
sources:
99
- https://github.com/osodevops/strimzi-backup-operator

src/controllers/backup.rs

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
use std::sync::Arc;
22

33
use futures::StreamExt;
4+
use k8s_openapi::api::batch::v1::Job;
45
use kube::{
56
runtime::{
67
controller::{Action, Controller},
@@ -46,6 +47,7 @@ fn error_policy(
4647

4748
pub async fn run(client: Client, metrics: Arc<MetricsState>) {
4849
let backups = Api::<KafkaBackup>::all(client.clone());
50+
let jobs = Api::<Job>::all(client.clone());
4951

5052
let context = Arc::new(Context {
5153
client: client.clone(),
@@ -55,6 +57,12 @@ pub async fn run(client: Client, metrics: Arc<MetricsState>) {
5557
info!("Starting KafkaBackup controller");
5658

5759
Controller::new(backups, Config::default().any_semantic())
60+
// Watch owned Jobs so completion/failure updates the KafkaBackup
61+
// status immediately instead of waiting for the periodic requeue.
62+
.owns(
63+
jobs,
64+
Config::default().labels("kafkabackup.com/type=backup"),
65+
)
5866
.shutdown_on_signal()
5967
.run(reconcile, error_policy, context)
6068
.for_each(|res| async move {

src/controllers/restore.rs

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
use std::sync::Arc;
22

33
use futures::StreamExt;
4+
use k8s_openapi::api::batch::v1::Job;
45
use kube::{
56
runtime::{
67
controller::{Action, Controller},
@@ -46,6 +47,7 @@ fn error_policy(
4647

4748
pub async fn run(client: Client, metrics: Arc<MetricsState>) {
4849
let restores = Api::<KafkaRestore>::all(client.clone());
50+
let jobs = Api::<Job>::all(client.clone());
4951

5052
let context = Arc::new(Context {
5153
client: client.clone(),
@@ -55,6 +57,12 @@ pub async fn run(client: Client, metrics: Arc<MetricsState>) {
5557
info!("Starting KafkaRestore controller");
5658

5759
Controller::new(restores, Config::default().any_semantic())
60+
// Watch owned Jobs so completion/failure updates the KafkaRestore
61+
// status immediately instead of waiting for the periodic requeue.
62+
.owns(
63+
jobs,
64+
Config::default().labels("kafkabackup.com/type=restore"),
65+
)
5866
.shutdown_on_signal()
5967
.run(reconcile, error_policy, context)
6068
.for_each(|res| async move {

src/jobs/job_state.rs

Lines changed: 88 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,88 @@
1+
use k8s_openapi::api::batch::v1::Job;
2+
3+
/// Aggregate state of the Jobs spawned for a backup/restore CR.
4+
///
5+
/// Derived from the full list of Jobs matching the CR's label selector so
6+
/// the reconciler can decide whether another Job may be created. A completed
7+
/// or failed Job must block re-creation just like a running one: backup and
8+
/// restore runs are one-shot, and pod-level retries are owned by the Job's
9+
/// `backoffLimit`, not the reconciler.
10+
#[derive(Debug, Clone, PartialEq, Eq)]
11+
pub enum JobsState {
12+
/// No Jobs exist for this resource.
13+
NoJobs,
14+
/// At least one Job is running or has not reached a terminal state yet.
15+
InProgress,
16+
/// A Job completed successfully.
17+
Succeeded { job_name: String },
18+
/// A Job failed terminally (exhausted its backoffLimit).
19+
Failed { job_name: String },
20+
}
21+
22+
/// Classify the Jobs belonging to one CR into an aggregate state.
23+
///
24+
/// Precedence: Succeeded > InProgress > Failed. A succeeded Job means the
25+
/// operation completed even if a stray duplicate is still running; a failed
26+
/// Job with a newer active run (manual re-trigger) reports running.
27+
pub fn classify_jobs(jobs: &[Job]) -> JobsState {
28+
if let Some(job) = jobs.iter().find(|j| job_succeeded(j)) {
29+
return JobsState::Succeeded {
30+
job_name: job.metadata.name.clone().unwrap_or_default(),
31+
};
32+
}
33+
// Any non-terminal Job counts as in progress, including a just-created
34+
// Job whose status the Job controller has not populated yet.
35+
if jobs.iter().any(|j| !job_failed(j)) {
36+
return JobsState::InProgress;
37+
}
38+
if let Some(job) = jobs.iter().find(|j| job_failed(j)) {
39+
return JobsState::Failed {
40+
job_name: job.metadata.name.clone().unwrap_or_default(),
41+
};
42+
}
43+
JobsState::NoJobs
44+
}
45+
46+
/// A Job succeeded if it reports a `Complete=True` condition or any
47+
/// succeeded pods.
48+
pub fn job_succeeded(job: &Job) -> bool {
49+
let Some(status) = job.status.as_ref() else {
50+
return false;
51+
};
52+
status.succeeded.unwrap_or(0) > 0 || has_condition(job, "Complete")
53+
}
54+
55+
/// A Job failed terminally only when it reports a `Failed=True` condition
56+
/// (backoffLimit exhausted or deadline exceeded). `status.failed > 0` alone
57+
/// counts pod retries still owned by the Job controller and is not terminal.
58+
pub fn job_failed(job: &Job) -> bool {
59+
has_condition(job, "Failed")
60+
}
61+
62+
fn has_condition(job: &Job, condition_type: &str) -> bool {
63+
job.status
64+
.as_ref()
65+
.and_then(|s| s.conditions.as_ref())
66+
.is_some_and(|conds| {
67+
conds
68+
.iter()
69+
.any(|c| c.type_ == condition_type && c.status == "True")
70+
})
71+
}
72+
73+
/// Whether a restore Job may be created. Restores are strictly one-shot:
74+
/// any existing Job — running, succeeded, or failed — suppresses creation.
75+
pub fn should_create_restore_job(state: &JobsState) -> bool {
76+
matches!(state, JobsState::NoJobs)
77+
}
78+
79+
/// Whether a one-shot backup Job may be created. A manual trigger
80+
/// (`kafkabackup.com/trigger=now`) requests a fresh run, so it bypasses the
81+
/// "a terminal Job already exists" gate — but never stacks onto an active Job.
82+
pub fn should_create_backup_job(state: &JobsState, manually_triggered: bool) -> bool {
83+
match state {
84+
JobsState::InProgress => false,
85+
JobsState::NoJobs => true,
86+
JobsState::Succeeded { .. } | JobsState::Failed { .. } => manually_triggered,
87+
}
88+
}

src/jobs/mod.rs

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
pub mod backup_job;
22
pub mod cronjob;
3+
pub mod job_state;
34
pub mod restore_job;
45
pub mod templates;

0 commit comments

Comments
 (0)