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

#43063) * Double-check TaskInstance state if it differs from Executor. * Update airflow/jobs/backfill_job_runner.py * Update airflow/jobs/backfill_job_runner.py * Update airflow/jobs/backfill_job_runner.py * Update airflow/jobs/backfill_job_runner.py * Update airflow/jobs/backfill_job_runner.py * Update airflow/jobs/backfill_job_runner.py --------- Co-authored-by: Utkarsh Sharma <utkarsharma2@gmail.com>
apache · Dec 9, 2024 · 90d6332 · 90d6332
1 parent 99e713e
commit 90d6332
Showing 1 changed file with 11 additions and 0 deletions.
diff --git a/airflow/jobs/backfill_job_runner.py b/airflow/jobs/backfill_job_runner.py
@@ -309,6 +309,17 @@ 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(
+                    "In-memory TaskInstance state %s does not agree with executor state %s. Attempting to resolve by refreshing in-memory task instance from DB.",
+                    ti,
+                    state,
+                )
+                ti.refresh_from_db(session=session)
+
             if (
                 state in (TaskInstanceState.FAILED, TaskInstanceState.SUCCESS)
                 and ti.state in self.STATES_COUNT_AS_RUNNING