DatabricksRunNowOperator(deferrable=True) crashes in triggerer: AsyncToSync used inside running event loop (Airflow 3.0.6)
- Dominant language
- Python
- Stars
- 46.9k
- Forks
- 17.8k
- Avg merge
- 2d 9h
- Merged PRs (30d)
- 472
Description
### Apache Airflow version
3.0.6
### If "Other Airflow 2 version" selected, which one?
_No response_
### What happened?
Running `DatabricksRunNowOperator(deferrable=True)` (and this likely applies to `DatabricksSubmitRunOperator` too, since both share the same trigger/hook code) submits the Databricks job successfully and the task defers correctly (worker slot released). But the triggerer then crashes while polling job status, before the run ever reaches a terminal state:
```
ERROR - Trigger failed:
Traceback (most recent call last):
File ".../airflow/jobs/triggerer_job_runner.py", line 963, in cleanup_finished_triggers
result = details["task"].result()
File ".../airflow/jobs/triggerer_job_runner.py", line 1072, in run_trigger
async for event in trigger.run():
File ".../airflow/providers/databricks/triggers/databricks.py", line 90, in run
run_state = await self.hook.a_get_run_state(self.run_id)
File ".../airflow/providers/databricks/hooks/databricks.py", line 514, in a_get_run_state
response = await self._a_do_api_call(GET_RUN_ENDPOINT, json)
File ".../airflow/providers/databricks/hooks/databricks_base.py", line 713, in _a_do_api_call
url = self._endpoint_url(full_endpoint)
File ".../airflow/providers/databricks/hooks/databricks_base.py", line 623, in _endpoint_url
port = f":{self.databricks_conn.port}" if self.databricks_conn.port else ""
File ".../functools.py", line 998, in __get__
val = self.func(instance)
File ".../airflow/providers/databricks/hooks/databricks_base.py", line 142, in databricks_conn
return self.get_connection(self.databricks_conn_id)
File ".../airflow/hooks/base.py", line 64, in get_connection
conn = Connection.get_connection_from_secrets(conn_id)
File ".../airflow/models/connection.py", line 478, in get_connection_from_secrets
conn = TaskSDKConnection.get(conn_id=conn_id)
File ".../airflow/sdk/definitions/connection.py", line 144, in get
return _get_connection(conn_id)
File ".../airflow/sdk/execution_time/context.py", line 160, in _get_connection
msg = SUPERVISOR_COMMS.send(GetConnection(conn_id=conn_id))
File ".../airflow/jobs/triggerer_job_runner.py", line 740, in send
return async_to_sync(self.asend)(msg)
File ".../asgiref/sync.py", line 186, in __call__
raise RuntimeError(
RuntimeError: You cannot use AsyncToSync in the same thread as an async event loop - just await the async function directly.
```
Root cause: `DatabricksHook.databricks_conn` is a synchronous `@cached_property` (`hooks/databricks_base.py`) that fetches the connection via the TaskSDK's `SUPERVISOR_COMMS.send()`, which internally bridges to sync via `asgiref.async_to_sync`. The trigger's async polling path (`DatabricksExecutionTrigger.run()` -> `hook.a_get_run_state()` -> `_a_do_api_call()` -> `_endpoint_url()`) calls this same synchronous property, but this time from *inside* the triggerer's already-running asyncio event loop. `async_to_sync` detects the running loop and refuses to bridge, raising `RuntimeError` instead of fetching the connection.
Since the trigger constructs a fresh `DatabricksHook` instance (`DatabricksExecutionTrigger.__init__`), `databricks_conn` has never been cached beforehand, so the first (and every) access inside `run()` hits this failure — the deferred task can never successfully poll.
I verified this is not fixed in the latest released provider: `apache-airflow-providers-databricks==7.18.1`'s `hooks/databricks_base.py` still defines `databricks_conn` as a plain synchronous `@cached_property` and `_a_do_api_call`/`_endpoint_url` are structurally unchanged from 7.7.1 — so bumping the provider version alone does not resolve this.
This looks like the same class of bug as #53447 (BigQuery hit the identical `AsyncToSync`-in-running-loop error). A Databricks-specific fix was proposed in #55568 ("Implement async version of databricks_conn in BaseDatabricksHook") but it stalled and was auto-closed as stale without merging, despite multiple people confirming the bug is still present ("this is not solved, please reopen"). A more general core-level fix (#55799, follow-up #57154) exists for the `ExecutionAPISecretsBackend` fallback path, but targets Airflow core >= 3.1.0 - we're not yet in a position to confirm whether that alone resolves the Databricks hook's specific synchronous `cached_property` pattern, since `_endpoint_url` calls `self.databricks_conn` directly rather than going through an async-safe path.
### What you think should happen instead?
`DatabricksRunNowOperator`/`DatabricksSubmitRunOperator` with `deferrable=True` should be able to poll run status from the triggerer without crashing. Either:
- Reopen and land #55568 (or an equivalent), giving `BaseDatabricksHook` an async-safe connection-fetch path that `_a_do_api_call`/`_endpoint_url` use when running inside the triggerer, or
- Confirm the core-level fix (#57154, Airflow >= 3.1.0) actually resolves this specific code path, and document the minimum core version required for Databricks deferrable operators to work.
### How to reproduce
1. Airflow 3.0.6, `apache-airflow-providers-databricks==7.7.1` (also reproduces on 7.18.1's unchanged code path per the above).
2. Any DAG using `DatabricksRunNowOperator(deferrable=True, job_name=..., databricks_conn_id=...)`.
3. Trigger the DAG. The job submits successfully (log shows "Run submitted with run_id: ..." and "Pausing task as DEFERRED"), then the triggerer immediately crashes with the `AsyncToSync` `RuntimeError` above on its first status poll.
### Operating System
N/A (AWS MWAA managed Airflow 3.0.6 environment)
### Versions of Apache Airflow Providers
apache-airflow-providers-databricks==7.7.1 (also verified 7.18.1 has the same unchanged code path)
### Deployment
Amazon (AWS) MWAA
### Deployment details
_No response_
### Anything else?
Related: #53447, #55568, #55799, #57154
### 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
Reproduce the failure with the stated Airflow and Databricks provider versions, then inspect providers/databricks/hooks/databricks_base.py and providers/databricks/triggers/databricks.py, especially databricks_conn, _endpoint_url, _a_do_api_call, and DatabricksExecutionTrigger.run(). Determine whether the async connection path or the referenced core fix resolves this code path; done means deferrable Databricks operators can poll to a terminal state without the triggerer crashing.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- python
- Domain
- api, backend
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Active
- Clarity
- Mostly clear
- Newbie friendliness
- 52/100