Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Double-check TaskInstance state if it differs from the Executor state. #43063

Merged
merged 10 commits into from
Dec 9, 2024
Merged

Double-check TaskInstance state if it differs from the Executor state. #43063

Changes from 5 commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
543afb2
Double-check TaskInstance state if it differs from Executor.
antonysouthworth-halter Oct 16, 2024
28ee863
Merge branch 'v2-10-test' into fix/42991
antonysouthworth-halter Oct 20, 2024
b6bec84
Merge branch 'v2-10-test' into fix/42991
antonysouthworth-halter Dec 4, 2024
0e3d607
Merge branch 'v2-10-test' into fix/42991
utkarsharma2 Dec 9, 2024
814f0bc
Update airflow/jobs/backfill_job_runner.py
utkarsharma2 Dec 9, 2024
88a241f
Update airflow/jobs/backfill_job_runner.py
utkarsharma2 Dec 9, 2024
3b8b397
Update airflow/jobs/backfill_job_runner.py
utkarsharma2 Dec 9, 2024
9575361
Update airflow/jobs/backfill_job_runner.py
utkarsharma2 Dec 9, 2024
d4435a7
Update airflow/jobs/backfill_job_runner.py
utkarsharma2 Dec 9, 2024
6f49289
Update airflow/jobs/backfill_job_runner.py
utkarsharma2 Dec 9, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions airflow/jobs/backfill_job_runner.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -309,6 +309,15 @@ def _manage_executor_state(

self.log.debug("Executor state: %s task %s", state, ti)

if (
state in (TaskInstanceState.FAILED, TaskInstanceState.SUCCESS)
and ti.state in self.STATES_COUNT_AS_RUNNING
):
self.log.debug(
f"In-memory TaskInstance state {ti} does not agree with executor state {state}. Attempting to resolve by refreshing in-memory task instance from DB."
)
ti.refresh_from_db(session=session)

if (
state in (TaskInstanceState.FAILED, TaskInstanceState.SUCCESS)
and ti.state in self.STATES_COUNT_AS_RUNNING
Expand Down
Loading