Double-check TaskInstance state if it differs from Executor.

apache · Oct 16, 2024 · 543afb2 · 543afb2
1 parent 2425ee0
commit 543afb2
Showing 1 changed file with 7 additions and 0 deletions.
diff --git a/airflow/jobs/backfill_job_runner.py b/airflow/jobs/backfill_job_runner.py
@@ -309,6 +309,13 @@ 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