Enhance ResumableJobMixin.get_job_status with context for better job status tracking

[amoghrajesh]

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

What?

ResumableJobMixin.get_job_status had no access to context, making it impossible to handle backends where the remote resource carrying job state can disappear after completion. I encountered this problem when running Spark with Kubernetes driver pods which after a run get garbage collected, where without context an implementation cannot distinguish "SUCCEEDED but pod gone" from "FAILED but pod gone". The only workaround was deleteOnTermination=false, which accumulates pods indefinitely.

The same gap will also appear in another form in cases when a job is tracked by external IDs (e.g. EMR (cluster_id, step_id), Glue (job_name, run_id)).

Current behaviour

get_job_status(self, external_id) has no context, no access to task_store, no way to cache terminal status before the remote resource disappears.

Proposed change

Add context: Context as a second parameter to get_job_status throughout the interface.

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[uranusjr]

I think this needs a backcompat shim so an old spark provider can work with new task-sdk and vice versa.

[amoghrajesh]

I don't think there was a release for providers yet after previous iteration of task sdk changes (added in #67118)?

[ashb]

Provider release is happening today I think. Speak to @potiuk. We could possibly exclude Spark from that release wave?

[ashb]

Change looks okay (though the spark op doesn't seem to use it yet? Follow on pr to make it use this?)

[amoghrajesh]

@ashb spark isn't going to use it. And not every operator needs to use it, this is mainly for K8s use case I mentioned in PR desc above

[amoghrajesh]

Error seems to be unrelated:

  ERROR: all download attempts failed for https://public.dhe.ibm.com/ibmdl/export/pub/software/websphere/messaging/mqdev/redist/9.4.0.0-IBM-MQC-Redist-LinuxX64.tar.gz; last error: <urlopen error [Errno 110] Connection timed out>
  Downloading https://public.dhe.ibm.com/ibmdl/export/pub/software/websphere/messaging/mqdev/redist/9.4.0.0-IBM-MQC-Redist-LinuxX64.tar.gz
  Primary download failed (URLError: <urlopen error [Errno 110] Connection timed out>); trying 4 fallback IP(s)
    Retrying with public.dhe.ibm.com -> 170.225.126.18
    170.225.126.18 failed: URLError: <urlopen error [Errno 110] Connection timed out>
    Retrying with public.dhe.ibm.com -> 129.35.224.1
    129.35.224.1 failed: URLError: <urlopen error [Errno 110] Connection timed out>
    Retrying with public.dhe.ibm.com -> 129.124.168.5
    129.124.168.5 failed: URLError: <urlopen error [Errno 110] Connection timed out>
    Retrying with public.dhe.ibm.com -> 9.133.44.11
    9.133.44.11 failed: URLError: <urlopen error [Errno 110] Connection timed out>
  Pre-extras install failed for ibm.mq

Merging this one in