feat(controller): emit JobSetRecreating event and add upgrade integration test

- In Reconcile(), detect the transient state where reconcileObjects()
  succeeds but the JobSet is NotFound and TrainJob is not suspended:
  this is the cycle after Build() deleted a stale suspended JobSet due
  to immutable spec.replicatedJobs change (RHOAIENG-48867). Record a
  Normal "JobSetRecreating" Kubernetes Event so operators can observe
  the automatic recreation via `kubectl describe trainjob`.

- Add integration test "Should delete and recreate stale suspended
  JobSet when runtime image changes post-upgrade" in
  test/integration/controller/trainjob_controller_test.go, covering
  the full end-to-end scenario: pre-upgrade state (suspended TrainJob
  + suspended JobSet with old image) → runtime image update →
  TrainJob resumed → JobSet deleted and recreated with new image.

Fixes RHOAIENG-48867

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: Ambient Code Bot

pkg/controller/trainjob_controller.go

@@ -26,6 +26,7 @@ import (
 	"github.com/go-logr/logr"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/equality"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/tools/record"
@@ -137,6 +138,15 @@ func (r *TrainJobReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 	setSuspendedCondition(&trainJob)
 
 	if statusErr := setTrainJobStatus(ctx, runtime, &trainJob); statusErr != nil {
+		// RHOAIENG-48867: When Build() deleted a stale suspended JobSet due to an
+		// immutable spec.replicatedJobs change (e.g., post-upgrade image update), the
+		// JobSet is transiently absent. Detect this by: reconcileObjects succeeded (err==nil),
+		// the TrainJob is not suspended, and the JobSet is NotFound. Record a Normal event
+		// so operators can observe the automatic recreation.
+		if err == nil && apierrors.IsNotFound(statusErr) && !ptr.Deref(trainJob.Spec.Suspend, false) {
+			r.recorder.Event(&trainJob, corev1.EventTypeNormal, "JobSetRecreating",
+				"Stale suspended JobSet deleted and will be recreated due to runtime spec changes post-upgrade")
+		}
 		err = errors.Join(err, statusErr)
 	}
 

test/integration/controller/trainjob_controller_test.go

@@ -1436,4 +1436,103 @@ alpha-node-0-1.alpha slots=8
 			})
 		})
 	})
+
+	// RHOAIENG-48867: Integration test for the upgrade scenario where a suspended
+	// TrainJob's ClusterTrainingRuntime image changes, requiring a delete-recreate
+	// of the immutable spec.replicatedJobs field in the JobSet.
+	ginkgo.When("Reconciling TrainJob after runtime image upgrade (RHOAIENG-48867)", func() {
+		var (
+			clTrainingRuntime *trainer.ClusterTrainingRuntime
+			trainJob          *trainer.TrainJob
+			trainJobKey       client.ObjectKey
+		)
+
+		const (
+			preUpgradeImage  = "registry.example.com/trainer:pre-upgrade"
+			postUpgradeImage = "registry.example.com/trainer:post-upgrade"
+		)
+
+		ginkgo.AfterEach(func() {
+			gomega.Expect(k8sClient.DeleteAllOf(ctx, &trainer.TrainJob{}, client.InNamespace(ns.Name))).Should(gomega.Succeed())
+			gomega.Expect(k8sClient.DeleteAllOf(ctx, &trainer.ClusterTrainingRuntime{})).Should(gomega.Succeed())
+		})
+
+		ginkgo.BeforeEach(func() {
+			clTrainingRuntime = testingutil.MakeClusterTrainingRuntimeWrapper("upgrade-runtime").
+				RuntimeSpec(
+					testingutil.MakeTrainingRuntimeSpecWrapper(
+						testingutil.MakeClusterTrainingRuntimeWrapper("upgrade-runtime").Spec,
+					).
+						Container(constants.Node, constants.Node, preUpgradeImage, nil, nil, nil).
+						Obj(),
+				).
+				Obj()
+
+			trainJob = testingutil.MakeTrainJobWrapper(ns.Name, "upgrade-trainjob").
+				Suspend(true).
+				RuntimeRef(trainer.GroupVersion.WithKind(trainer.ClusterTrainingRuntimeKind), clTrainingRuntime.Name).
+				Obj()
+			trainJobKey = client.ObjectKeyFromObject(trainJob)
+		})
+
+		ginkgo.It("Should delete and recreate stale suspended JobSet when runtime image changes post-upgrade", func() {
+			ginkgo.By("Creating ClusterTrainingRuntime with pre-upgrade image")
+			gomega.Expect(k8sClient.Create(ctx, clTrainingRuntime)).Should(gomega.Succeed())
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(clTrainingRuntime), clTrainingRuntime)).Should(gomega.Succeed())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+			ginkgo.By("Creating suspended TrainJob (simulating pre-upgrade state)")
+			gomega.Expect(k8sClient.Create(ctx, trainJob)).Should(gomega.Succeed())
+
+			ginkgo.By("Waiting for suspended JobSet to be created with pre-upgrade image")
+			gomega.Eventually(func(g gomega.Gomega) {
+				jobSet := &jobsetv1alpha2.JobSet{}
+				g.Expect(k8sClient.Get(ctx, trainJobKey, jobSet)).Should(gomega.Succeed())
+				g.Expect(ptr.Deref(jobSet.Spec.Suspend, false)).Should(gomega.BeTrue())
+				for _, rJob := range jobSet.Spec.ReplicatedJobs {
+					if rJob.Name == constants.Node {
+						for _, c := range rJob.Template.Spec.Template.Spec.Containers {
+							if c.Name == constants.Node {
+								g.Expect(c.Image).Should(gomega.Equal(preUpgradeImage))
+							}
+						}
+					}
+				}
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+			ginkgo.By("Simulating upgrade: updating ClusterTrainingRuntime with post-upgrade image")
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(clTrainingRuntime), clTrainingRuntime)).Should(gomega.Succeed())
+				clTrainingRuntime.Spec = testingutil.MakeTrainingRuntimeSpecWrapper(clTrainingRuntime.Spec).
+					Container(constants.Node, constants.Node, postUpgradeImage, nil, nil, nil).
+					Obj()
+				g.Expect(k8sClient.Update(ctx, clTrainingRuntime)).Should(gomega.Succeed())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+			ginkgo.By("Resuming TrainJob post-upgrade (simulating Kueue admitting the workload)")
+			gomega.Eventually(func(g gomega.Gomega) {
+				g.Expect(k8sClient.Get(ctx, trainJobKey, trainJob)).Should(gomega.Succeed())
+				trainJob.Spec.Suspend = ptr.To(false)
+				g.Expect(k8sClient.Update(ctx, trainJob)).Should(gomega.Succeed())
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+			ginkgo.By("Verifying the JobSet is recreated with the post-upgrade image and is not suspended")
+			gomega.Eventually(func(g gomega.Gomega) {
+				jobSet := &jobsetv1alpha2.JobSet{}
+				g.Expect(k8sClient.Get(ctx, trainJobKey, jobSet)).Should(gomega.Succeed())
+				g.Expect(ptr.Deref(jobSet.Spec.Suspend, false)).Should(gomega.BeFalse())
+				for _, rJob := range jobSet.Spec.ReplicatedJobs {
+					if rJob.Name == constants.Node {
+						for _, c := range rJob.Template.Spec.Template.Spec.Containers {
+							if c.Name == constants.Node {
+								g.Expect(c.Image).Should(gomega.Equal(postUpgradeImage),
+									"JobSet should have post-upgrade image after delete-recreate")
+							}
+						}
+					}
+				}
+			}, util.Timeout, util.Interval).Should(gomega.Succeed())
+		})
+	})
 })