You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by po...@apache.org on 2026/09/21 21:52:50 UTC
(airflow) branch main updated: Record TaskInstanceHistory when a task fails from a non-running state (#67372)
This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 3b29a4c3b3d Record TaskInstanceHistory when a task fails from a non-running state (#67372)
3b29a4c3b3d is described below
commit 3b29a4c3b3d26142b80dad1c10481f449da5ecaa
Author: Pradeep Kalluri <12...@users.noreply.github.com>
AuthorDate: Mon Sep 21 22:52:39 2026 +0100
Record TaskInstanceHistory when a task fails from a non-running state (#67372)
When a task instance failed from a non-RUNNING state (QUEUED or SCHEDULED after
being killed externally by the executor), handle_failure() set UP_FOR_RETRY without
calling prepare_db_for_next_try(), so no TaskInstanceHistory row was written and the
attempt lost its hostname and execution metadata. The UI then had no log entry for
that try.
The condition now excludes only RESTARTING, whose history is already recorded when a
running task is cleared.
closes: #65366
closes: #67238
---
airflow-core/newsfragments/67372.bugfix.rst | 1 +
airflow-core/src/airflow/models/taskinstance.py | 12 +++--
airflow-core/tests/unit/jobs/test_scheduler_job.py | 51 ++++++++++++++++++++++
3 files changed, 60 insertions(+), 4 deletions(-)
diff --git a/airflow-core/newsfragments/67372.bugfix.rst b/airflow-core/newsfragments/67372.bugfix.rst
new file mode 100644
index 00000000000..3cd5c2f6368
--- /dev/null
+++ b/airflow-core/newsfragments/67372.bugfix.rst
@@ -0,0 +1 @@
+Fix TaskInstanceHistory missing host/metadata when task is retried from a non-RUNNING state (e.g. killed externally or failed in queued/deferred status)
diff --git a/airflow-core/src/airflow/models/taskinstance.py b/airflow-core/src/airflow/models/taskinstance.py
index 5c8559d11f4..d9c3f8cab95 100644
--- a/airflow-core/src/airflow/models/taskinstance.py
+++ b/airflow-core/src/airflow/models/taskinstance.py
@@ -1933,10 +1933,14 @@ class TaskInstance(Base, LoggingMixin, BaseWorkload):
if task and fail_fast:
_stop_remaining_tasks(task_instance=ti, session=session)
else:
- if ti.state == TaskInstanceState.RUNNING:
- # If the task instance is in the running state, it means it raised an exception and
- # about to retry so we record the task instance history. For other states, the task
- # instance was cleared and already recorded in the task instance history.
+ if ti.state != TaskInstanceState.RESTARTING:
+ # Record the current attempt and prepare the TI for its next try.
+ # Covers every path eligible for retry reaching handle_failure():
+ # - RUNNING: task raised an exception during execution (normal failure)
+ # - QUEUED/SCHEDULED: executor killed the task externally before
+ # it could start (e.g. pod OOMKilled in KubernetesExecutor)
+ # RESTARTING is excluded: the task was cleared via the UI/API while running;
+ # prepare_db_for_next_try() was already called during that clear operation.
ti.prepare_db_for_next_try(session)
ti.state = State.UP_FOR_RETRY
diff --git a/airflow-core/tests/unit/jobs/test_scheduler_job.py b/airflow-core/tests/unit/jobs/test_scheduler_job.py
index d3ed94e70e8..05d6a76c7cc 100644
--- a/airflow-core/tests/unit/jobs/test_scheduler_job.py
+++ b/airflow-core/tests/unit/jobs/test_scheduler_job.py
@@ -5427,6 +5427,57 @@ class TestSchedulerJob:
is not None
)
+ @pytest.mark.parametrize("ti_state", [TaskInstanceState.QUEUED, TaskInstanceState.SCHEDULED])
+ def test_process_executor_events_queued_ti_retry_preserves_history(self, ti_state, dag_maker, session):
+ """
+ Regression test for #65366 / #67238.
+
+ When an executor reports FAILED for a TI that is QUEUED or SCHEDULED
+ (killed externally before it could start), the scheduler calls handle_failure()
+ which must call prepare_db_for_next_try() so that TaskInstanceHistory is recorded
+ with the correct hostname and start_date.
+ """
+ dag_id = "test_queued_ti_retry_history"
+ task_id = "dummy"
+ hostname = "worker-node-42"
+
+ with dag_maker(dag_id=dag_id, fileloc="/test_path/"):
+ task = EmptyOperator(task_id=task_id, retries=2)
+
+ dr = dag_maker.create_dagrun()
+ ti = dr.get_task_instance(task.task_id, session=session)
+ ti.state = ti_state
+ ti.hostname = hostname
+ ti.start_date = DEFAULT_DATE
+ ti.try_number = 1
+ ti.max_tries = 2
+ session.merge(ti)
+ session.commit()
+
+ old_ti_id = ti.id
+
+ executor = MockExecutor(do_update=False)
+ executor.event_buffer[ti.key] = TaskInstanceState.FAILED, None
+
+ scheduler_job = Job()
+ self.job_runner = SchedulerJobRunner(job=scheduler_job, executors=[executor])
+ self.job_runner._process_executor_events(executor=executor, session=session)
+
+ session.expire_all()
+ ti.refresh_from_db(session=session)
+
+ assert ti.state == State.UP_FOR_RETRY
+ assert ti.id != old_ti_id, "prepare_db_for_next_try must assign a new UUID"
+
+ from airflow.models.taskinstancehistory import TaskInstanceHistory
+
+ tih = session.scalar(
+ select(TaskInstanceHistory).where(TaskInstanceHistory.task_instance_id == old_ti_id)
+ )
+ assert tih is not None, "TaskInstanceHistory must be created for non-RUNNING retry"
+ assert tih.hostname == hostname
+ assert tih.start_date == DEFAULT_DATE
+
def test_adopt_or_reset_orphaned_tasks_external_triggered_dag(self, dag_maker, session):
dag_id = "test_reset_orphaned_tasks_external_triggered_dag"
with dag_maker(dag_id=dag_id, schedule="@daily"):