anish-heidi opened a new issue, #71525:
URL: https://github.com/apache/airflow/issues/71525
### 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)
--
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]
