fix: respect cleanPodPolicy when job exceeds backoffLimit

Move UpdateJobConditions(JobFailed) before DeletePodsAndServices in
the jobExceedsLimit block so that IsFinished() returns true and the
cleanPodPolicy: None guard is not bypassed.

Previously, DeletePodsAndServices was called before the JobFailed
condition was set, causing all pods to be unconditionally deleted
regardless of cleanPodPolicy when a job exceeded its backoffLimit.

Tested on live cluster: pods now preserved after backoffLimit failure
with cleanPodPolicy: None.

Fixes: #3419
Signed-off-by: Aviad Hayumi <aviad.hayumi@run.ai>

Author: AviadHayumi

pkg/controller.v1/common/job.go

@@ -220,8 +220,12 @@ func (jc *JobController) ReconcileJobs(
 			jobStatus.CompletionTime = &now
 		}
 
-		// If the Job exceeds backoff limit or is past active deadline
-		// delete all pods and services, then set the status to failed
+		// Set JobFailed condition BEFORE cleanup so that DeletePodsAndServices
+		// sees IsFinished()==true and correctly respects cleanPodPolicy.
+		// Fix: https://github.com/kubeflow/trainer/issues/3419
+		jc.Recorder.Event(runtimeObject, corev1.EventTypeNormal, commonutil.NewReason(jobKind, commonutil.JobFailedReason), failureMessage)
+		commonutil.UpdateJobConditions(&jobStatus, apiv1.JobFailed, corev1.ConditionTrue, commonutil.NewReason(jobKind, commonutil.JobFailedReason), failureMessage)
+
 		if err := jc.DeletePodsAndServices(runtimeObject, runPolicy, jobStatus, pods); err != nil {
 			return err
 		}
@@ -240,10 +244,6 @@ func (jc *JobController) ReconcileJobs(
 			}
 		}
 
-		jc.Recorder.Event(runtimeObject, corev1.EventTypeNormal, commonutil.NewReason(jobKind, commonutil.JobFailedReason), failureMessage)
-
-		commonutil.UpdateJobConditions(&jobStatus, apiv1.JobFailed, corev1.ConditionTrue, commonutil.NewReason(jobKind, commonutil.JobFailedReason), failureMessage)
-
 		return jc.Controller.UpdateJobStatusInApiServer(job, &jobStatus)
 	} else {
 		// General cases which need to reconcile

pkg/controller.v1/common/job_test.go

@@ -97,6 +97,16 @@ func TestDeletePodsAndServices(T *testing.T) {
 			wantPods:       &corev1.PodList{},
 			wantService:    &corev1.ServiceList{},
 		},
+		// This test reproduces the backoffLimit bug: when DeletePodsAndServices is called
+		// without JobFailed condition set (as happens in the jobExceedsLimit path before
+		// the fix), pods are deleted despite cleanPodPolicy: None.
+		// See: https://github.com/kubeflow/trainer/issues/3419
+		"Unfinished job with cleanPodPolicy None deletes pods (pre-fix backoffLimit bug)": {
+			cleanPodPolicy: apiv1.CleanPodPolicyNone,
+			jobCondition:   "",
+			wantPods:       &corev1.PodList{},
+			wantService:    &corev1.ServiceList{},
+		},
 	}
 	for name, tc := range cases {
 		T.Run(name, func(t *testing.T) {