seanmuth opened a new issue, #51903:
URL: https://github.com/apache/airflow/issues/51903
### Apache Airflow version
2.11.0
### If "Other Airflow 2 version" selected, which one?
_No response_
### What happened?
Task for a DAG is in the running state. During that time, the DAG is removed
from the DAGs dir.
When the scheduler goes to mark the task as a zombie due to LTJ not
heartbeating after completion, it's unable to:
```
[2025-06-18T18:32:37.054+0000] {processor.py:211} ERROR - Got an exception!
Propagating...
Traceback (most recent call last):
File
"/usr/local/lib/python3.12/site-packages/airflow/dag_processing/processor.py",
line 199, in _run_file_processor
_handle_dag_file_processing()
File
"/usr/local/lib/python3.12/site-packages/airflow/dag_processing/processor.py",
line 188, in _handle_dag_file_processing
result: tuple[int, int, int] = dag_file_processor.process_file(
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.12/site-packages/airflow/utils/session.py",
line 97, in wrapper
return func(*args, session=session, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File
"/usr/local/lib/python3.12/site-packages/airflow/dag_processing/processor.py",
line 937, in process_file
DagFileProcessor.execute_callbacks_without_dag(callback_requests,
self.UNIT_TEST_MODE)
File
"/usr/local/lib/python3.12/site-packages/airflow/api_internal/internal_api_call.py",
line 166, in wrapper
return func(*args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.12/site-packages/airflow/utils/session.py",
line 97, in wrapper
return func(*args, session=session, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File
"/usr/local/lib/python3.12/site-packages/airflow/dag_processing/processor.py",
line 786, in execute_callbacks_without_dag
cls._execute_task_callbacks(None, request, unit_test_mode, session)
File
"/usr/local/lib/python3.12/site-packages/airflow/api_internal/internal_api_call.py",
line 166, in wrapper
return func(*args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.12/site-packages/airflow/utils/session.py",
line 94, in wrapper
return func(*args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^
File
"/usr/local/lib/python3.12/site-packages/airflow/dag_processing/processor.py",
line 875, in _execute_task_callbacks
ti.handle_failure(error=request.msg, test_mode=unit_test_mode,
session=session)
File "/usr/local/lib/python3.12/site-packages/airflow/utils/session.py",
line 94, in wrapper
return func(*args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^
File
"/usr/local/lib/python3.12/site-packages/airflow/models/taskinstance.py", line
3457, in handle_failure
_handle_failure(
File "/usr/local/lib/python3.12/site-packages/airflow/utils/session.py",
line 94, in wrapper
return func(*args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^
File
"/usr/local/lib/python3.12/site-packages/airflow/api_internal/internal_api_call.py",
line 166, in wrapper
return func(*args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^
File
"/usr/local/lib/python3.12/site-packages/airflow/models/taskinstance.py", line
1232, in _handle_failure
failure_context = TaskInstance.fetch_handle_failure_context(
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File
"/usr/local/lib/python3.12/site-packages/airflow/models/taskinstance.py", line
3406, in fetch_handle_failure_context
get_listener_manager().hook.on_task_instance_failed(
File "/usr/local/lib/python3.12/site-packages/pluggy/_hooks.py", line 513,
in __call__
return self._hookexec(self.name, self._hookimpls.copy(), kwargs,
firstresult)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.12/site-packages/pluggy/_manager.py", line
120, in _hookexec
return self._inner_hookexec(hook_name, methods, kwargs, firstresult)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.12/site-packages/pluggy/_callers.py", line
139, in _multicall
raise exception.with_traceback(exception.__traceback__)
File "/usr/local/lib/python3.12/site-packages/pluggy/_callers.py", line
103, in _multicall
res = hook_impl.function(*args)
^^^^^^^^^^^^^^^^^^^^^^^^^
File
"/usr/local/lib/python3.12/site-packages/airflow/providers/openlineage/plugins/listener.py",
line 384, in on_task_instance_failed
self._on_task_instance_failed(task_instance, task.dag,
task_instance.dag_run, task, error)
^^^^^^^^
AttributeError: 'NoneType' object has no attribute 'dag'
```
this results in the scheduler attempting to mark this TI as a Zombie during
_every_ scheduler loop, generating tons of false positive zombie metrics.
### What you think should happen instead?
Fail task gracefully, don't attempt to execute callbacks because they're
gone.
### How to reproduce
1. Start dag with a running task
2. While task is running, remove DAG from dags dir
### Operating System
n/a
### Versions of Apache Airflow Providers
n/a
### Deployment
Astronomer
### Deployment details
Runtime 13.0.0 / Airflow 2.11.0+astro.1
Celery Executor
### Anything else?
_No response_
### Are you willing to submit PR?
- [x] Yes I am willing to submit a PR!
### Code of Conduct
- [x] I agree to follow this project's [Code of
Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
--
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]
