Add job_postprocessor hook to Kubernetes Job launchers

[morgan-wowk]

Is it safe to deploy?

Yes — this is fully backward compatible. job_postprocessor is a new keyword-only parameter defaulting to None, so every existing launcher call site is unaffected and the Job-transform hook stays a no-op until a caller explicitly opts in.

What

Adds a job_postprocessor hook to the Kubernetes Job launchers, symmetric to the existing pod_postprocessor. It lets a caller transform the V1Job object right before it's created, threaded through _KubernetesJobLauncher, _KubernetesPodOrJobLauncher, and both GoogleKubernetesEngine_UsingGoogleCloudStorage_* launchers.

class JobPostProcessor(typing.Protocol):
    def __call__(
        self, *, job: k8s_client_lib.V1Job, annotations: dict[str, str] | None = None
    ) -> k8s_client_lib.V1Job: ...

The default _transform_job_before_launching now applies the injected job_postprocessor when one is set, and is otherwise the same no-op it was before.

Why

The launchers already let callers customize the Pod (via pod_postprocessor), and for multi-node runs that Pod becomes the Job's pod template — so pod-level changes already reach Job pods. But there was no way to set fields on the V1Job itself without editing this repo:

• _transform_job_before_launching was a no-op override hook living on _KubernetesJobLauncher, but the class callers actually build is _KubernetesPodOrJobLauncher, which constructs a private inner _KubernetesJobLauncher with no hook threaded through. So subclassing the public launcher can't reach it.

The concrete need: a downstream (Tangle/oasis) execution cluster with no Kueue and a janitor that holds no batch/jobs permission, so finished Job objects are never reaped. Stamping spec.ttlSecondsAfterFinished lets Kubernetes' native TTL-after-finished controller clean them up — but that field is on the Job, not the pod template, so pod_postprocessor can't set it. With this hook the caller does:

def _set_job_ttl(*, job, annotations=None):
    if job.spec.ttl_seconds_after_finished is None:
        job.spec.ttl_seconds_after_finished = 7 * 24 * 60 * 60
    return job

launcher = GoogleKubernetesEngine_..._KubernetesPodOrJobLauncher(
    ..., job_postprocessor=_set_job_ttl,
)

System impact

• No behavior change when unused. job_postprocessor defaults to None; the transform hook stays a no-op, so every existing launch path is byte-identical. New keyword-only param with a default → no call-site is required to change.
• Applies only to the Job path (multi-node runs); the Pod path is untouched.

Before / after

• Before: the only way to influence the launched Job was to fork/subclass the inner _KubernetesJobLauncher, which public callers don't instantiate → effectively not extensible.
• After: callers pass a job_postprocessor the same way they already pass pod_postprocessor.

Tests

tests/test_kubernetes_launchers.py (new, offline — no live K8s API or GCS):

• _transform_job_before_launching applies the job_postprocessor when set;
• it's a no-op (returns the same Job unchanged) when unset;
• _KubernetesPodOrJobLauncher forwards job_postprocessor to its inner Job launcher.

All pass; black-formatted.