apache / apache/airflow

DatabricksRunNowOperator(deferrable=True) crashes in triggerer: AsyncToSync used inside running event loop (Airflow 3.0.6)

Open
#71,525 12 comments 0 reactions 0 assignees View on GitHub
area:async-operators area:providers good first issue kind:bug provider:databricks
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.