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 <noreply@anthropic.com>

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
 
 

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
+    try:
-    max_attempts = settings.confirm_max_retries + 1
+        result = run_spark_submit(cmd)
-    for attempt in range(1, max_attempts + 1):
+    except SparkSubmitError as exc:
-        try:
+        pending.status = "FAILED"
-            result = run_spark_submit(cmd)
+        pending.error = str(exc)
-        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
         pending_store.save(pending)
-
+        logger.error(
-        job_store.put(
+            f"confirm_submit_job failed pending_id={pending_id} err={exc}"
-            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,
-            )
         )
+        raise
 
-        logger.info(
+    application_id, tracking_url = parse_spark_submit_output(result.stderr)
-            f"confirm_submit_job ok pending_id={pending_id} job_id={job_id} "
+    job_id = uuid.uuid4().hex[:12]
-            f"application_id={application_id}"
+
-        )
+    # Persist pending as SUBMITTED before creating the in-memory Job so a
-        return SubmitResult(
+    # 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]]:

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, match="boom"):
-    with pytest.raises(SparkSubmitError):
         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 ---
+# --- confirm idempotency / manual retry ---
-
-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 "")
-
 
 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):
+def test_confirm_resets_failed_then_succeeds(monkeypatch, real_script):
-    monkeypatch.setattr(time, "sleep", lambda _s: None)
+    """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