ImportError: cannot import name 'SUPERVISOR_COMMS' with dag.test()
- Dominant language
- Python
- Stars
- 46.9k
- Forks
- 17.8k
- Avg merge
- 2d 10h
- Merged PRs (30d)
- 483
Description
### Apache Airflow version
3.0.2
### If "Other Airflow 2 version" selected, which one?
_No response_
### What happened?
The exception occurred when the DAG ran with `dag.test()` attempted to retrieve a variable from the API server. Some similar issues have been opened (#48554, #51062, #51316). The PRs provided as a solution (#50300, #50419) were included in 3.0.2 but did not fix the problem.
```
Exception has occurred: ImportError
cannot import name 'SUPERVISOR_COMMS' from 'airflow.sdk.execution_time.task_runner' (/workspaces/airflow/.venv/lib/python3.12/site-packages/airflow/sdk/execution_time/task_runner.py)
File "/workspaces/airflow/test.py", line 7, in
x = Variable.get("my_variable")
^^^^^^^^^^^^^^^^^^^^^^^^
ImportError: cannot import name 'SUPERVISOR_COMMS' from 'airflow.sdk.execution_time.task_runner' (/workspaces/airflow/.venv/lib/python3.12/site-packages/airflow/sdk/execution_time/task_runner.py)
```
### What you think should happen instead?
The variable should have been successfully retrieved without exceptions.
### How to reproduce
1. Set the variable: `airflow variables set my_variable my_value`
2. Run DAG:
```
import logging
from airflow import DAG
from airflow.providers.standard.operators.empty import EmptyOperator
from airflow.providers.standard.operators.python import PythonOperator
from airflow.sdk import Variable
x = Variable.get("my_variable")
def my_function(my_var: str) -> None:
logging.getLogger(__name__).info(my_var)
with DAG("test_dag") as dag:
start = EmptyOperator(task_id="start")
py_func = PythonOperator(
task_id="py_func",
python_callable=my_function,
op_kwargs={
"my_var": x
}
)
end = EmptyOperator(task_id="end")
start >> py_func >> end
if __name__ == "__main__":
dag.test()
```
### Operating System
Debian GNU/Linux 12 (bookworm)
### Versions of Apache Airflow Providers
_No response_
### Deployment
Other Docker-based deployment
### Deployment details
Extended image based on `apache/airflow:slim-3.0.2-python3.12`
### Anything else?
_No response_
### Are you willing to submit PR?
- [ ] 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)
Contributor guide
Research direction
Start with the reproducer in test.py, focusing on Variable.get("my_variable") during dag.test(), and trace the ImportError for SUPERVISOR_COMMS. Review related issues #48554, #51062, and #51316 plus PRs #50300 and #50419; done means the shown DAG retrieves the variable without an ImportError on Airflow 3.0.2.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- python
- Domain
- data-engineering
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 38/100