[Karman580]

What this PR does / why we need it:

Problem: SparkClient.wait_for_job_status() does not check the requested status against SparkJobStatus. If a value can never match the job's status, the Kubernetes backend keeps polling until timeout instead of reporting invalid input. Two cases:

• A misspelled status such as {"Complete"}, which is the Trainer/Optimizer spelling. Spark uses "Completed".
• An empty set.

Solution: KubernetesBackend.wait_for_job_status() now validates the requested statuses against set(SparkJobStatus) before polling. It raises ValueError if the collection is empty or contains a value that is not a valid SparkJobStatus.

• This mirrors the existing status.issubset(...) checks in the Trainer and Optimizer Kubernetes backends.
• The check uses set(SparkJobStatus) rather than a hard-coded list, so it stays correct if enum members are added, as fix(spark): treat Spark Operator retry states as non-terminal #690 proposes.
• The Raises: docs in the backend, the RuntimeBackend base interface, and SparkClient are updated to match.

Compatibility:

• SparkJobStatus is a str enum, so enum members and correctly spelled plain strings such as {"Completed"} behave as before.
• The status: set[SparkJobStatus] signature and its {SparkJobStatus.COMPLETED} default are unchanged.
• The only inputs that now fail fast are ones that previously could never return successfully:
  • invalid values
  • an empty set
  • a bare string that is not an exact single status, such as "Complete". A bare str is outside the set[...] type hint.

Tests (run locally on macOS):

• Added cases to the parametrized test_wait_for_job_status in kubeflow/spark/backends/kubernetes/backend_test.py:

  • {"Complete"} raises ValueError
  • set() raises ValueError
  • {"Completed"} (plain string) returns the job

  The existing enum case {SparkJobStatus.COMPLETED} is unchanged and still passes.

• uv run pytest -q kubeflow/spark/backends/kubernetes/backend_test.py -k wait_for_job_status: 10 passed.

• With the backend change reverted, the two new ValueError cases fail. That confirms they cover the defect.

• I ran the commands behind make verify directly: uv lock --check, ruff check, ruff format --check kubeflow, ty check kubeflow/hub. All passed. The Makefile's GNU sed check stops make itself on macOS.

• I ran the test command behind make test-python directly: uv run coverage run --source=kubeflow -m pytest ./kubeflow/. 700 passed.

• The pre-commit hooks passed on the changed files.

Note: #690 also modifies wait_for_job_status, in the docstring and the RuntimeError message. The overlap is small, and I am happy to rebase on whichever lands first.

Which issue(s) this PR fixes:

Fixes #873

Checklist:

• Docs included if any changes are user facing

[google-oss-prow]

[APPROVALNOTIFIER] This PR is NOT APPROVED

This pull-request has been approved by:
Once this PR has been reviewed and has the lgtm label, please assign kramaranya for approval. For more information see the Kubernetes Code Review Process.

The full list of commands accepted by this bot can be found here.

Details

Needs approval from an approver in each of these files:

• OWNERS

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

[tariq-hasan]

/ok-to-test
/retest

[Karman580]

Hi @tariq-hasan, @szaher ,@Shekharrajak could you please take a look at my PR #874 and review it when you get a chance

[tariq-hasan]

No need for the comments. The code is self-explanatory.

Suggested change

	# An invalid or empty status can never match, so fail fast instead of polling until timeout.
	valid_statuses = set(SparkJobStatus)
	# A bare str is iterable, so set("Completed") would yield characters rather than a status.
	if isinstance(status, str):
	raise ValueError(
	f"status must be a set of SparkJobStatus values, e.g. "
	f"{{SparkJobStatus.COMPLETED}}; got {status!r}"
	)
	valid_statuses = set(SparkJobStatus)
	if isinstance(status, str):
	raise ValueError(
	f"status must be a set of SparkJobStatus values, e.g. "
	f"{{SparkJobStatus.COMPLETED}}; got {status!r}"
	)

[Karman580]

Thanks for taking another look, @tariq-hasan! You're right, the code reads fine without them. I've removed both comments and applied your suggestion as-is in 91c5857, and the tests still pass locally.

Whenever you have a moment, I'd really appreciate another review. Thanks again for your patience and the helpful feedback!

[Karman580]

Hi @tariq-hasan, a quick heads-up on the failing example-validation (1.32.0) check. It doesn't look related to this change. The notebook passes a valid {SparkJobStatus.COMPLETED}, so the new validation lets it through, but the Spark job itself goes to FAILED in the cluster a few seconds after submission.

It seems to be the controller/Helm chart mismatch described in #933 (the same failure shows up on #910), which #930 should take care of. Once that lands, I'll rebase this branch on main so the checks run again on the fixed setup.

Thanks a lot for your time!