Integrations whose stop is another controller's work still read as stopped too early

[thc1006]

What is wrong:

Three integrations still finish their Workload, and release its quota, from a status their own controller has not caught up with. If that controller then acts on its earlier view, Pods exist against quota that has already been given away.

#14441 added JobWithStopAcknowledgement so the framework can wait for the controller that actually carries out a stop, rather than finishing the Workload as soon as the stop is submitted. Where it landed:

Integration	State
batch/Job	implemented, JobSuspended condition
JobSet	implemented, Suspended condition
RayCluster	implemented, RayClusterSuspended bound to the reported generation
RayService	implemented, both cluster slots suspended
Kubeflow (PyTorchJob, TFJob, PaddleJob, XGBoostJob, JAXJob)	implemented once, in the shared KubeflowJob wrapper
TrainJob	implemented, Suspended condition
AppWrapper	not needed: IsActive already reads QuotaReserved, a condition its controller writes rather than a count
Pod	not needed: Pod.Stop deletes the Pods itself
RayJob	open
MPIJob	open
SparkApplication	open

The three that are open have no signal in their pinned upstream API to implement it against:

• RayJob (pkg/controller/jobs/rayjob/rayjob_controller.go:108) is inactive in New as well as Suspended. New is the zero value, so "KubeRay has not seen this yet" and "nothing was ever created" cannot be told apart, and KubeRay v1.6.2 writes status.observedGeneration for RayCluster and RayService only.
• MPIJob declares JobSuspended in v2beta1/types.go but mpi-operator v0.8.2 never writes it. On suspend it writes JobRunning=False with a reason constant that is not exported.
• SparkApplication (pkg/controller/jobs/sparkapplication/sparkapplication_controller.go:84) is active only in Running. SparkApplicationStatus carries neither conditions nor observedGeneration, so there is nothing to read. Its own IsActive also treats Submitted as inactive while the driver may already exist.

Why is this needed:

The path that reaches it is a prebuilt Workload that does not match its Job: Kueue stops the Job and finishes the Workload as OutOfSync, which releases the quota. That is the cross-controller form of what #14395 described within one controller.

This is not a regression from #14441. Before it there was no wait at all here, so every integration is better off than it was.

Two things this issue should not be read as asking for:

The default cannot simply be reversed. stopAcknowledged returning false ends the reconcile through errWaitingForStop, which returns without requeueing. An integration whose controller will never report a suspend then holds its Workload unfinished and its quota unreleased for good. That is not hypothetical: on a MultiKueue manager the training operator skips a Job whose managedBy points elsewhere (pytorchjob_controller.go:135), the adapter only mirrors the remote status, and MultiKueue deletes remote objects rather than suspending them, so JobSuspended never arrives.

And what the implemented ones report is what each parent controller says about itself. How far that reaches into the Job's own children varies: JobSet sets its condition in the same call that writes spec.suspend on the child Jobs, without waiting for the Job controllers; the Trainer controller applies the child JobSet and then sets its condition from the parent spec. Answering the child chain needs a check that can take a context and list owned resources, which a zero-argument bool cannot. #13308 is already reworking IsActive into IsActive(ctx), and that looks like the right place for it.

Completion requirements:

• RayJob, once KubeRay reports something that separates "not yet observed" from "nothing created"
• MPIJob, once mpi-operator writes the JobSuspended it already declares
• SparkApplication, once its status carries conditions or an observed generation
• Decide whether observation should move to a context-taking check that can inspect owned children, alongside Impl/quota release wiring #13308's IsActive(ctx)
• Decide whether an integration with a synchronous stop should say so explicitly, and whether the default can then be reversed

---

This issue was written in part with the assistance of generative AI.

[thc1006]

/kind cleanup
/area integrations

[thc1006]

/remove-kind cleanup
/kind bug

Retitling the kind here rather than leaving it as cleanup. What this describes is a quota overcommit: the Workload is finished and its capacity handed on while the controller that owns the Job has not yet observed the stop and can still create Pods from its earlier view. That is a correctness problem for the integrations listed, not deferred tidying.