1fanwang opened a new pull request, #72417:
URL: https://github.com/apache/airflow/pull/72417
A synchronous Dataproc job can outlive its worker, but a task retry
currently submits a second job instead of reconnecting. After this change, the
retry restores the API-returned job ID from task state and resumes waiting for
the same job.
`DataprocSubmitJobOperator` uses the AIP-103 resumable-job contract only on
its synchronous wait path. Active jobs reconnect, completed jobs return without
another submission, and terminal or missing jobs submit fresh. Recovery
restores the operator job ID and extra link state used by cancellation.
Asynchronous, deferrable, direct-to-trigger, OpenLineage, and pre-Airflow 3.3
behavior stay unchanged.
An explicit `request_id` is still passed unchanged. Dataproc may therefore
return the prior terminal job when a retry resubmits with the same idempotency
key.
## Testing
```console
$ PATH="$PWD/.venv/bin:$PATH"
AIRFLOW_HOME="$PWD/.build/airflow-home-dataproc-final" uv run --project
providers/google pytest
providers/google/tests/unit/google/cloud/operators/test_dataproc.py -q
125 passed, 1 warning in 97.29s
```
A scratch harness executed two operator instances against a local Dataproc
hook and the real `TaskStateStoreAccessor` path:
```console
$ BASE="$PWD/.build/dataproc-baseline";
AIRFLOW_HOME="$PWD/.build/airflow-home-dataproc-proof-baseline"
PYTHONPATH="$BASE/airflow-core/src:$BASE/task-sdk/src:$BASE/providers/common/compat/src:$BASE/providers/google/src"
"$PWD/.venv/bin/python" dev/dataproc_resumability_e2e.py --label baseline
$ AIRFLOW_HOME="$PWD/.build/airflow-home-dataproc-proof-patched"
PYTHONPATH="$PWD/airflow-core/src:$PWD/task-sdk/src:$PWD/providers/common/compat/src:$PWD/providers/google/src"
"$PWD/.venv/bin/python" dev/dataproc_resumability_e2e.py --label patched
```
<details><summary>Raw crash/retry output</summary>
```json
{"label":"baseline","error":"WorkerCrash: simulated worker crash while
waiting for Dataproc
job","state":{},"submissions":["dataproc-job-1","dataproc-job-2"],"recovered":false}
{"label":"patched","error":"WorkerCrash: simulated worker crash while
waiting for Dataproc
job","state":{"dataproc_job_id":"dataproc-job-1"},"submissions":["dataproc-job-1"],"recovered":true}
```
</details>
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes - GitHub Copilot CLI (GPT-5.6 Sol)
Generated-by: GitHub Copilot CLI (GPT-5.6 Sol) following [the
guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
---
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
