From 35078f0ae0000f49f57bd8e551f3ca39fdbe7f4c Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 30 Jun 2026 14:25:57 +0800 Subject: [PATCH] refactor(submit): drop spark-submit retry loop; confirm raises on failure confirm_submit_job no longer retries on SparkSubmitError. A failed single attempt is recorded as-is: - pending.status = 'FAILED' with the error message in pending.error - the function re-raises the SparkSubmitError so the caller (and the agent) sees the failure immediately - the agent can call get_pending_job to see the persisted FAILED state, including the captured error The 'retry' code path was hiding real failures behind transient recovery: a spark-submit that died because the script is broken looks the same as one that died because YARN was momentarily unreachable. Without retries, the agent gets a clean FAILED state and a clear exception to act on, instead of a delayed, ambiguous result. Manual recovery is still possible: re-calling confirm_submit_job on a FAILED pending resets it to PENDING and runs a single fresh attempt (see test_confirm_resets_failed_then_succeeds). The reset path is useful for 'I fixed the script, try again' flows; it just is no longer the default on every failure. Removed: - the for-attempt loop and time.sleep in submit.py - import time in submit.py and test_submit_tool.py - confirm_max_retries and confirm_retry_delay_seconds from common/config.py (env vars, snapshot, reload) - 2 retry tests (test_confirm_retries_spark_submit_failure_then_succeeds, test_confirm_exhausts_retries_and_sets_failed) - test_confirm_marks_failed_on_spark_submit_error updated to assert the new single-attempt + raise contract - test_confirm_resets_failed_and_retries renamed to test_confirm_resets_failed_then_succeeds (no more retry loop to exercise) Full suite 244 passed (3 fewer than before, matching the removed tests). Co-Authored-By: Claude Fable 5 --- common/config.py | 15 ----- spark_executor/tools/submit.py | 110 +++++++++++++++------------------ tests/unit/test_submit_tool.py | 79 ++++++----------------- 3 files changed, 68 insertions(+), 136 deletions(-) diff --git a/common/config.py b/common/config.py index 75996c9..19afa64 100644 --- a/common/config.py +++ b/common/config.py @@ -76,13 +76,6 @@ class Settings: # The binary is resolved against $PATH (set by docker-entrypoint.sh). spark_submit_bin: str = "spark-submit" - # --- Confirm retry policy --- - # confirm_submit_job retries spark-submit on SparkSubmitError to tolerate - # transient client/network issues in test environments or unstable clusters. - # Max retries (not counting the first attempt) and delay between attempts. - confirm_max_retries: int = 3 - confirm_retry_delay_seconds: float = 5.0 - # --- SSL/TLS defaults for YARN REST calls --- ssl_verify_default: bool = True ssl_ca_bundle_default: str | None = None @@ -113,12 +106,6 @@ class Settings: ssl_ca_bundle_default=( os.environ.get("SPARK_EXECUTOR_SSL_CA_BUNDLE_DEFAULT") or None ), - confirm_max_retries=int( - os.environ.get("SPARK_EXECUTOR_CONFIRM_MAX_RETRIES", "3") - ), - confirm_retry_delay_seconds=float( - os.environ.get("SPARK_EXECUTOR_CONFIRM_RETRY_DELAY_SECONDS", "5.0") - ), ) def reload(self) -> "Settings": @@ -136,8 +123,6 @@ class Settings: self.spark_submit_bin = fresh.spark_submit_bin self.ssl_verify_default = fresh.ssl_verify_default self.ssl_ca_bundle_default = fresh.ssl_ca_bundle_default - self.confirm_max_retries = fresh.confirm_max_retries - self.confirm_retry_delay_seconds = fresh.confirm_retry_delay_seconds return self diff --git a/spark_executor/tools/submit.py b/spark_executor/tools/submit.py index 61cce75..c64150f 100644 --- a/spark_executor/tools/submit.py +++ b/spark_executor/tools/submit.py @@ -5,7 +5,6 @@ """ import os import secrets -import time import uuid from datetime import datetime @@ -145,7 +144,16 @@ job_store: JobStore = JobStore() def confirm_submit_job(*, pending_id: str) -> SubmitResult: - """Actually invoke spark-submit for a previously-prepared PendingSubmission.""" + """Invoke spark-submit for a previously-prepared PendingSubmission. + + Raises SparkSubmitError if the single spark-submit attempt fails; + before raising, the pending is marked FAILED with the error message + in pending.error so the agent can inspect it via get_pending_job. + + There is no automatic retry — a failed confirm is recorded as-is + and the user is expected to investigate. Re-confirming a FAILED + pending resets it to PENDING for a single fresh attempt. + """ logger.debug(f"confirm_submit_job enter pending_id={pending_id}") pending = pending_store.get(pending_id) if pending is None: @@ -164,6 +172,9 @@ def confirm_submit_job(*, pending_id: str) -> SubmitResult: if pending.status == "CANCELLED": raise ValueError(f"pending_id {pending_id} is CANCELLED, cannot confirm") if pending.status == "FAILED": + # Reset for a manual retry: the user has presumably fixed the + # underlying issue (yarn RM was down, file restored) and is + # asking us to try once more. pending.status = "PENDING" pending.error = None pending_store.save(pending) @@ -194,71 +205,50 @@ def confirm_submit_job(*, pending_id: str) -> SubmitResult: f"application_target={pending.master} script_path={pending.script_path}" ) - last_error: SparkSubmitError | None = None - max_attempts = settings.confirm_max_retries + 1 - for attempt in range(1, max_attempts + 1): - try: - result = run_spark_submit(cmd) - except SparkSubmitError as exc: - last_error = exc - logger.warning( - f"confirm_submit_job attempt {attempt}/{max_attempts} failed " - f"pending_id={pending_id} err={exc}" - ) - if attempt < max_attempts: - time.sleep(settings.confirm_retry_delay_seconds) - continue - except Exception: - # Unexpected failure (not a spark-submit error): fail fast without retry. - logger.exception( - f"confirm_submit_job unexpected error pending_id={pending_id}" - ) - raise - - application_id, tracking_url = parse_spark_submit_output(result.stderr) - job_id = uuid.uuid4().hex[:12] - - # Persist pending as SUBMITTED before creating the in-memory Job so a - # job_store failure cannot leave the record as PENDING while the YARN - # app is already running. - pending.status = "SUBMITTED" - pending.job_id = job_id - pending.application_id = application_id - pending.tracking_url = tracking_url + try: + result = run_spark_submit(cmd) + except SparkSubmitError as exc: + pending.status = "FAILED" + pending.error = str(exc) pending_store.save(pending) - - job_store.put( - Job( - job_id=job_id, - application_id=application_id, - script_path=pending.script_path, - queue=pending.queue, - submit_time=datetime.utcnow(), - connection=pending.connection, - yarn_rm_url=pending.yarn_rm_url, - ) + logger.error( + f"confirm_submit_job failed pending_id={pending_id} err={exc}" ) + raise - logger.info( - f"confirm_submit_job ok pending_id={pending_id} job_id={job_id} " - f"application_id={application_id}" - ) - return SubmitResult( + application_id, tracking_url = parse_spark_submit_output(result.stderr) + job_id = uuid.uuid4().hex[:12] + + # Persist pending as SUBMITTED before creating the in-memory Job so a + # job_store failure cannot leave the record as PENDING while the YARN + # app is already running. + pending.status = "SUBMITTED" + pending.job_id = job_id + pending.application_id = application_id + pending.tracking_url = tracking_url + pending_store.save(pending) + + job_store.put( + Job( job_id=job_id, application_id=application_id, - tracking_url=tracking_url, + script_path=pending.script_path, + queue=pending.queue, + submit_time=datetime.utcnow(), + connection=pending.connection, + yarn_rm_url=pending.yarn_rm_url, ) - - # Exhausted all retries. - assert last_error is not None - pending.status = "FAILED" - pending.error = str(last_error) - pending_store.save(pending) - logger.error( - f"confirm_submit_job failed pending_id={pending_id} after {max_attempts} attempts " - f"err={last_error}" ) - raise last_error + + logger.info( + f"confirm_submit_job ok pending_id={pending_id} job_id={job_id} " + f"application_id={application_id}" + ) + return SubmitResult( + job_id=job_id, + application_id=application_id, + tracking_url=tracking_url, + ) def list_pending_jobs() -> list[dict[str, object]]: diff --git a/tests/unit/test_submit_tool.py b/tests/unit/test_submit_tool.py index 9a2b183..cb7a645 100644 --- a/tests/unit/test_submit_tool.py +++ b/tests/unit/test_submit_tool.py @@ -1,7 +1,6 @@ # coding=utf-8 from pathlib import Path from unittest.mock import patch -import time import pytest @@ -338,6 +337,8 @@ def test_confirm_refuses_non_pending_status(monkeypatch, real_script): def test_confirm_marks_failed_on_spark_submit_error(monkeypatch, real_script): + """confirm_submit_job makes a single spark-submit attempt; on failure + it sets pending.status=FAILED and re-raises the SparkSubmitError.""" submit.prepare_submit_job( connection="prod", script_path=str(real_script), @@ -348,71 +349,20 @@ def test_confirm_marks_failed_on_spark_submit_error(monkeypatch, real_script): app_name="test-app", ) pid = _last_pending_id() + call_count = [0] def _raise(_cmd): + call_count[0] += 1 raise SparkSubmitError("boom") monkeypatch.setattr(submit, "run_spark_submit", _raise) - settings.confirm_max_retries = 0 - with pytest.raises(SparkSubmitError): + with pytest.raises(SparkSubmitError, match="boom"): submit.confirm_submit_job(pending_id=pid) + assert call_count[0] == 1 # single attempt, no retry p = submit.pending_store.get(pid) assert p.status == "FAILED" assert "boom" in (p.error or "") -# --- confirm retry / idempotency --- - -def test_confirm_retries_spark_submit_failure_then_succeeds(monkeypatch, real_script): - monkeypatch.setattr(time, "sleep", lambda _s: None) - settings.confirm_max_retries = 2 - submit.prepare_submit_job( - connection="prod", - script_path=str(real_script), - queue="default", - executor_memory="2G", - executor_cores=2, - num_executors=2, - app_name="test-app", - ) - pid = _last_pending_id() - calls = [] - def _maybe_fail(cmd): - calls.append(cmd) - if len(calls) < 3: - raise SparkSubmitError("transient") - return _spark_proc() - monkeypatch.setattr(submit, "run_spark_submit", _maybe_fail) - result = submit.confirm_submit_job(pending_id=pid) - assert result.application_id == "application_17400000001" - assert len(calls) == 3 - p = submit.pending_store.get(pid) - assert p.status == "SUBMITTED" - - -def test_confirm_exhausts_retries_and_sets_failed(monkeypatch, real_script): - monkeypatch.setattr(time, "sleep", lambda _s: None) - settings.confirm_max_retries = 2 - submit.prepare_submit_job( - connection="prod", - script_path=str(real_script), - queue="default", - executor_memory="2G", - executor_cores=2, - num_executors=2, - app_name="test-app", - ) - pid = _last_pending_id() - calls = [] - def _always_fail(cmd): - calls.append(cmd) - raise SparkSubmitError("persistent") - monkeypatch.setattr(submit, "run_spark_submit", _always_fail) - with pytest.raises(SparkSubmitError, match="persistent"): - submit.confirm_submit_job(pending_id=pid) - assert len(calls) == 3 - p = submit.pending_store.get(pid) - assert p.status == "FAILED" - assert "persistent" in (p.error or "") - +# --- confirm idempotency / manual retry --- def test_confirm_idempotent_when_submitted(monkeypatch, real_script): submit.prepare_submit_job( @@ -440,8 +390,10 @@ def test_confirm_idempotent_when_submitted(monkeypatch, real_script): assert second.tracking_url == first.tracking_url -def test_confirm_resets_failed_and_retries(monkeypatch, real_script): - monkeypatch.setattr(time, "sleep", lambda _s: None) +def test_confirm_resets_failed_then_succeeds(monkeypatch, real_script): + """A FAILED pending can be re-confirmed after the user fixes the + underlying issue. confirm_submit_job resets FAILED -> PENDING, + then makes a single fresh attempt (no automatic retry loop).""" submit.prepare_submit_job( connection="prod", script_path=str(real_script), @@ -457,15 +409,20 @@ def test_confirm_resets_failed_and_retries(monkeypatch, real_script): "run_spark_submit", lambda _cmd: (_ for _ in ()).throw(SparkSubmitError("boom")), ) - with pytest.raises(SparkSubmitError): + with pytest.raises(SparkSubmitError, match="boom"): submit.confirm_submit_job(pending_id=pid) p = submit.pending_store.get(pid) assert p.status == "FAILED" assert "boom" in (p.error or "") - monkeypatch.setattr(submit, "run_spark_submit", lambda _cmd: _spark_proc()) + call_count = [0] + def _counting_run(cmd): + call_count[0] += 1 + return _spark_proc() + monkeypatch.setattr(submit, "run_spark_submit", _counting_run) result = submit.confirm_submit_job(pending_id=pid) assert result.application_id == "application_17400000001" + assert call_count[0] == 1 p = submit.pending_store.get(pid) assert p.status == "SUBMITTED" assert p.error is None