From 9305ac750b185d95a62a66ea927487c526dbefe6 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 30 Jun 2026 14:54:44 +0800 Subject: [PATCH] fix(submit): create Job record for failed submits with recovered application_id MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Audit of the previous 'salvage application_id from stderr' change (e4fb97c) revealed the fix was incomplete: - The pending record now had pending.application_id set on failure. - But get_job_logs / get_job_status / kill_job all look up the application_id through JobStore.get_either(application_id), which scans the on-disk jobs.json for a Job record matching that application_id. No Job record was being created on the failure path, so the recovered application_id was unreachable through the public tool surface — get_job_logs would 404 with 'No Job found for id=...application_xxx'. Fix: when confirm_submit_job's failure branch successfully extracts an application_id from stderr, it now also creates a Job record (same shape as the success path: 12-char hex job_id, application_id, script_path, queue, submit_time, connection, yarn_rm_url). The Job's only job in this case is to act as the lookup index from the LLM-facing tools into the underlying YARN app. The job_id is generated even when application_id recovery fails (so we always set pending.job_id), but the Job record is only put when application_id is non-None. That keeps the 'spark-submit died locally before reaching YARN' case clean: no Job record, get_job_logs gets the standard 'no logs available' error from yarn_client, not a 404 from JobStore. Tests: - test_confirm_recovers_application_id_from_failed_spark_submit now asserts both that pending.application_id is set AND that a Job record exists and is findable by both job_id AND application_id. - test_confirm_failed_spark_submit_without_application_id_keeps_it_none asserts the negative case: no application_id, no Job record. Fixture fix: _fresh() now rebinds submit.job_store to use tmp_path. Previously the test wrote to ./data/jobs.json (the real data dir), which left stale records between test runs and broke any test that scanned by application_id. This is a latent bug exposed by the new assertions; fixing it here also makes every other confirm test hermetic. Co-Authored-By: Claude Fable 5 --- spark_executor/tools/submit.py | 36 +++++++++++++++++++++++++++++----- tests/unit/test_submit_tool.py | 30 +++++++++++++++++++++++++--- 2 files changed, 58 insertions(+), 8 deletions(-) diff --git a/spark_executor/tools/submit.py b/spark_executor/tools/submit.py index f8014ff..10ff531 100644 --- a/spark_executor/tools/submit.py +++ b/spark_executor/tools/submit.py @@ -212,10 +212,12 @@ def confirm_submit_job(*, pending_id: str) -> SubmitResult: # auth expired, the YARN app was launched but spark-submit lost # its connection, the PySpark script exited non-zero after # YARN had already accepted it, ...). In several of those cases - # the stderr still contains "Submitted application " — try - # to salvage it so the user can fetch logs via - # get_application_logs on the failed job. Failure to recover - # application_id is non-fatal; we still mark FAILED and raise. + # the stderr still contains "Submitted application ". + # We must salvage it AND create a Job record keyed by a fresh + # job_id, so the user can fetch logs / status / kill the + # underlying YARN app via the normal get_job_* tools (which + # all look up through JobStore). Without a Job record the + # application_id is just a string with no log target. if getattr(exc, "result", None) is not None: try: application_id, tracking_url = parse_spark_submit_output( @@ -226,15 +228,39 @@ def confirm_submit_job(*, pending_id: str) -> SubmitResult: else: application_id, tracking_url = None, None + job_id = uuid.uuid4().hex[:12] pending.status = "FAILED" pending.error = str(exc) + pending.job_id = job_id if application_id: pending.application_id = application_id pending.tracking_url = tracking_url pending_store.save(pending) + + if application_id: + # Create a Job record so get_job_logs / get_job_status / + # kill_job can find the application_id. The Job's + # application_id is the only field that matters for those + # lookups; everything else is best-effort metadata. + 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.info( + f"confirm_submit_job recorded Job for failed submit: " + f"job_id={job_id} application_id={application_id}" + ) + logger.error( f"confirm_submit_job failed pending_id={pending_id} " - f"application_id={application_id} err={exc}" + f"job_id={job_id} application_id={application_id} err={exc}" ) raise diff --git a/tests/unit/test_submit_tool.py b/tests/unit/test_submit_tool.py index f1b15c1..5c74ae9 100644 --- a/tests/unit/test_submit_tool.py +++ b/tests/unit/test_submit_tool.py @@ -6,6 +6,7 @@ import pytest from common.config import settings from spark_executor.core import connection_store, pending_store +from spark_executor.core.job_store import JobStore from spark_executor.core.pending_store import PendingStore from spark_executor.models import Connection from spark_executor.tools import submit @@ -22,6 +23,10 @@ def _fresh(tmp_path: Path, monkeypatch): monkeypatch.setattr(connection_store, "store", connection_store.ConnectionStore()) monkeypatch.setattr(pending_store, "DEFAULT_DATA_DIR", str(tmp_path)) monkeypatch.setattr(pending_store, "store", PendingStore()) + # submit.job_store is a module-level singleton bound at import time + # to settings.data_dir (./data/). Rebind to tmp_path so failed-submit + # tests don't pollute the real data/jobs.json with old records. + submit.job_store = JobStore(data_dir=str(tmp_path)) submit.conn_store = connection_store.store submit.pending_store = pending_store.store connection_store.store.save(Connection( @@ -367,8 +372,9 @@ def test_confirm_recovers_application_id_from_failed_spark_submit(monkeypatch, r """When spark-submit fails AFTER YARN accepted the application (a common case: spark-submit loses its connection to the RM, or the PySpark script exits non-zero on the cluster side), the stderr - still contains 'Submitted application '. We must salvage it so - the user can call get_application_logs on the failed job.""" + still contains 'Submitted application '. We must salvage it + AND create a Job record so get_job_logs / get_job_status / + kill_job can find the application_id through JobStore.""" submit.prepare_submit_job( connection="prod", script_path=str(real_script), @@ -396,10 +402,24 @@ def test_confirm_recovers_application_id_from_failed_spark_submit(monkeypatch, r p = submit.pending_store.get(pid) assert p.status == "FAILED" assert "lost connection to RM" in (p.error or "") - # The whole point of this test: application_id must be set even on failure. assert p.application_id == "application_17400000002_0007" assert p.tracking_url == "http://rm:8088/proxy/application_17400000002_0007/" + # The Job record must exist so get_job_logs / get_job_status / + # kill_job can find the application_id through JobStore. The + # agent only sees the exception; the persistent log target lives + # in JobStore. + assert p.job_id is not None + job = submit.job_store.get_either(p.job_id) + assert job is not None + assert job.application_id == "application_17400000002_0007" + # And the application_id lookup must also work (this is the path + # an LLM agent typically takes — they have the YARN id, not the + # local job_id). + job_by_app = submit.job_store.get_either("application_17400000002_0007") + assert job_by_app is not None + assert job_by_app.job_id == p.job_id + def test_confirm_failed_spark_submit_without_application_id_keeps_it_none(monkeypatch, real_script): """spark-submit can fail BEFORE reaching YARN (e.g. local script @@ -432,6 +452,10 @@ def test_confirm_failed_spark_submit_without_application_id_keeps_it_none(monkey assert p.status == "FAILED" assert p.application_id is None assert p.tracking_url is None + # No application_id → no Job record. get_either for the pending's + # job_id must return None. + if p.job_id is not None: + assert submit.job_store.get_either(p.job_id) is None # --- confirm idempotency / manual retry ---