Do not create the launcher job if the job starts suspended

Author: GonzaloSaez

pkg/controller/mpi_job_controller.go

@@ -658,7 +658,9 @@ func (c *MPIJobController) syncHandler(key string) error {
 				return err
 			}
 		}
-		if launcher == nil {
+		// If the job is suspended, the list of worker pods will be incorrect. We also do
+		// not want to start the launcher job if the MPIJob starts suspended.
+		if launcher == nil && !isMPIJobSuspended(mpiJob) {
 			if mpiJob.Spec.LauncherCreationPolicy == kubeflow.LauncherCreationPolicyAtStartup || c.countReadyWorkerPods(worker) == len(worker) {
 				launcher, err = c.kubeClient.BatchV1().Jobs(namespace).Create(context.TODO(), c.newLauncherJob(mpiJob), metav1.CreateOptions{})
 				if err != nil {

test/integration/mpi_job_controller_test.go

@@ -173,6 +173,27 @@ func TestMPIJobSuccess(t *testing.T) {
 }
 
 func TestMPIJobWaitWorkers(t *testing.T) {
+	testcases := []struct {
+		name           string
+		startSuspended bool
+	}{
+		{
+			name:           "don't start suspended",
+			startSuspended: false,
+		},
+		{
+			name:           "start suspended",
+			startSuspended: true,
+		},
+	}
+	for _, tc := range testcases {
+		t.Run(tc.name, func(t *testing.T) {
+			testMpiJobWaitWorkers(t, tc.startSuspended)
+		})
+	}
+}
+
+func testMpiJobWaitWorkers(t *testing.T, startSuspended bool) {
 	ctx, cancel := context.WithCancel(context.Background())
 	t.Cleanup(cancel)
 	s := newTestSetup(ctx, t)
@@ -187,6 +208,7 @@ func TestMPIJobWaitWorkers(t *testing.T) {
 			SlotsPerWorker:         ptr.To[int32](1),
 			LauncherCreationPolicy: "WaitForWorkersReady",
 			RunPolicy: kubeflow.RunPolicy{
+				Suspend:        ptr.To(startSuspended),
 				CleanPodPolicy: ptr.To(kubeflow.CleanPodPolicyRunning),
 			},
 			MPIReplicaSpecs: map[kubeflow.MPIReplicaType]*kubeflow.ReplicaSpec{