1fanwang opened a new pull request, #72423:
URL: https://github.com/apache/airflow/pull/72423
A synchronous `DbtCloudRunJobOperator` submits a duplicate dbt Cloud run
when its worker crashes while polling.
Before this change, a retry starts another run. After it, the task stores
its exact run ID through AIP-103, reconnects to active runs, restores
successful runs, and submits fresh after errors or cancellations. Recovery
applies only to synchronous waits and never searches for another active run.
Existing deferrable, no-wait, reuse, and retry-from-failure behavior is
unchanged.
## Testing
The full provider suite passed `281` tests, including `18` focused recovery
tests. Pre-commit and manual checks passed. A local stand-in hook drove the
real `execute()` task-state path below; the scratch script stayed uncommitted.
```bash
ROOT=$PWD
RED_HOME=$(mktemp -d /tmp/airflow-dbt-red.XXXXXX)
GREEN_HOME=$(mktemp -d /tmp/airflow-dbt-green.XXXXXX)
(cd .build/dbt-e2e-upstream && \
AIRFLOW_HOME="$RED_HOME" .venv/bin/python \
"$ROOT/dev/dbt_cloud_resumability_e2e.py" --expect red)
AIRFLOW_HOME="$GREEN_HOME" .venv/bin/python \
dev/dbt_cloud_resumability_e2e.py --expect green
```
<details>
<summary>Raw crash/retry logs</summary>
Upstream `main`:
```text
Waiting for job run 1001 to terminate.
{"error": "worker died while waiting for run 1001", "event": "worker_crash"}
{"event": "after_first_attempt", "submissions": [1001], "task_state": {}}
Waiting for job run 1002 to terminate.
Job run 1002 has completed successfully.
{"event": "after_retry", "exact_lookups": [], "result": 1002, "submissions":
[1001, 1002], "task_state": {}, "xcoms": {"job_run_id": 1002, "job_run_url":
"https://cloud.getdbt.com/deploy/1/projects/2/runs/1002/"}}
```
This branch:
```text
Waiting for job run 1001 to terminate.
{"error": "worker died while waiting for run 1001", "event": "worker_crash"}
{"event": "after_first_attempt", "submissions": [1001], "task_state":
{"dbt_cloud_run_id": 1001}}
Reconnecting to existing job external_id=1001
external_id_key=dbt_cloud_run_id status=RUNNING
Waiting for job run 1001 to terminate.
Job run 1001 has completed successfully.
{"event": "after_retry", "exact_lookups": [1001], "result": 1001,
"submissions": [1001], "task_state": {"dbt_cloud_run_id": 1001}, "xcoms":
{"job_run_id": 1001, "job_run_url":
"https://cloud.getdbt.com/deploy/1/projects/2/runs/1001/"}}
```
</details>
<!-- SPDX-License-Identifier: Apache-2.0
https://www.apache.org/licenses/LICENSE-2.0 -->
---
##### 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)
---
* Read the **[Pull Request
Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#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](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals))
is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License
Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment:
`{pr_number}.significant.rst`, in
[airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments).
You can add this file in a follow-up commit after the PR is created so you
know the PR number.
--
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]
