tao.chen/mcp-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(submit): create Job record for failed submits with recovered application_id

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
Claude
2026-06-30 14:54:44 +08:00
co-authored by Claude Fable 5
parent e4fb97c5cc
commit 9305ac750b
2 changed files with 58 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
spark_executor/tools/submit.py
+31 -5
View File
@@ -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 <id>" — 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 <id>".
# 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
tests/unit/test_submit_tool.py
+27 -3
View File
@@ -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 <id>'. We must salvage it so
the user can call get_application_logs on the failed job."""
still contains 'Submitted application <id>'. 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 ---