Remote XCOM: RuntimeError: This event loop is already running
- Dominant language
- Python
- Stars
- 46.9k
- Forks
- 17.8k
- Avg merge
- 2d 7h
- Merged PRs (30d)
- 484
Description
### Apache Airflow version
3.0.3
### If "Other Airflow 2 version" selected, which one?
_No response_
### What happened?
When I run kubernetes pod operators with XCOM:
```yaml
# Store XCOM in the xcom folder
- AIRFLOW__CORE__XCOM_BACKEND=airflow.providers.common.io.xcom.backend.XComObjectStorageBackend
- AIRFLOW__COMMON_IO__XCOM_OBJECTSTORAGE_PATH=abfs://wasb_default@/xcom
- AIRFLOW__COMMON_IO__XCOM_OBJECTSTORAGE_THRESHOLD=0
- AIRFLOW_CONN_WASB_DEFAULT=wasb://:@?extra__wasb__account_name=extra__wasb__account_key=
```
The task runs, the XCOM to Azure Blob Storage is written, but the **task fails with RuntimeError: This event loop is already running**:
```logs
RuntimeError: This event loop is already running
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/providers/cncf/kubernetes/operators/pod.py", line 680 in execute_sync
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/providers/cncf/kubernetes/operators/pod.py", line 611 in await_pod_start
File "/usr/local/lib/python3.12/asyncio/base_events.py", line 667 in run_until_complete
File "/usr/local/lib/python3.12/asyncio/base_events.py", line 626 in _check_running
```
Notes:
- When I run normal tasks (not KubernetesPodTasks) the run succeeds.
- The run succeeds when I turn the custom XCOM backend of (`AIRFLOW__COMMON_IO__XCOM_OBJECTSTORAGE_THRESHOLD=-1`).
### What you think should happen instead?
https://github.com/apache/airflow/blob/6f1243a6e5bba80a918ab7fc9f4dc2f6beee9330/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py#L609
```py
def await_pod_start(self, pod: k8s.V1Pod) -> None:
try:
async def _await_pod_start():
events_task = self.pod_manager.watch_pod_events(pod, self.startup_check_interval_seconds)
pod_start_task = self.pod_manager.await_pod_start(
pod=pod,
schedule_timeout=self.schedule_timeout_seconds,
startup_timeout=self.startup_timeout_seconds,
check_interval=self.startup_check_interval_seconds,
)
await asyncio.gather(pod_start_task, events_task)
asyncio.run(_await_pod_start()) # <- PROBLEM IS HERE
```
Calling asyncio.run(...) inside a context where an event loop is already active.
Possible solutions:
- `nest_asyncio` or
- `loop.run_until_complete(...)`
### How to reproduce
```py
from typing import List
from airflow.decorators import dag, task
from defaults_args import default_args, default_pod_args
# Common configuration for Kubernetes pods
default_pod_args = {
"image": "python:3.12-slim-bookworm",
"image_pull_policy": "IfNotPresent",
"get_logs": True,
"do_xcom_push": True,
},
# Common configuration for Airflow DAGs
default_args = {
"owner": "airflow",
"depends_on_past": False,
"email_on_failure": False,
"email_on_retry": False,
"retries": 0,
}
@dag(
dag_id="test",
schedule=None,
default_args=default_args,
)
def test():
@task.kubernetes(
task_id="test_list",
name="test_list",
**default_pod_args,
)
def make_index_list(count: int) -> List[int]:
return list(range(count))
@task.kubernetes(
task_id="test_expand",
name="test_expand",
**default_pod_args,
)
def hello(index):
print(f"hello: {index}")
indexes = make_index_list(3)
hello.expand(index=indexes)
pipeline = test()
```
### Operating System
Mac OS
### Versions of Apache Airflow Providers
I run the airflow container: `apache/airflow:3.0.3`
### Deployment
Official Apache Airflow Helm Chart
### Deployment details
_No response_
### 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 in providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py, especially execute_sync and await_pod_start, then run the supplied Kubernetes task DAG with the remote XCom backend targeting Azure Blob Storage. Done means the task completes successfully with remote XCom enabled, including when an event loop is already active.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- azure, kubernetes, python
- Domain
- backend, data-engineering
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 45/100