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

[anish-heidi]

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 Implement async version of databricks_conn in BaseDatabricksHook #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 (Fix connection access in triggerer for deferrable operators #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

• I agree to follow this project's Code of Conduct

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[anish-heidi]

Update: this is already resolved in Airflow core >= 3.2.1 (possibly earlier, haven't bisected precisely) - just not in a way that's obvious from the databricks provider alone, which is why I initially thought a databricks-specific fix was still needed.

Checked the source at the 3.2.1 tag directly:

• task-sdk/src/airflow/sdk/execution_time/secrets/execution_api.py's get_connection() now catches the RuntimeError from async_to_sync and falls back to greenback.await_(self.aget_connection(conn_id)) when called inside a greenback portal.
• airflow-core/src/airflow/jobs/triggerer_job_runner.py confirms every trigger runs inside such a portal (await greenback.ensure_portal()).

This is the general core-level fix from #55799 (merged 2025-09-18, landed in 3.1.0), regressed by #56602 (2025-10-14), and re-fixed by #57154 (2025-10-23). It patches the problem one layer below any individual hook/provider, in Airflow's own connection-fetch machinery - so DatabricksHook.databricks_conn's synchronous cached_property pattern (and presumably any other provider hook with the same pattern) works transparently once core is upgraded past the #57154 fix, without needing #55568 or any other databricks-specific patch.

For anyone hitting this: the fix is "upgrade Airflow core to >= 3.2.x", not "wait for a provider patch." Leaving this open since 3.0.x/3.1.0-3.1.x users without the #57154 fix would still hit it, but wanted to close the loop on my own findings here.

[eladkal]

cc @moomindani from Databricks team to share your thoughts here

[moomindani]

Thanks for the thorough write-up, and for coming back with your own follow-up analysis — that saved a lot of digging.

Your conclusion is right in substance: this is a core-level bug, it is fixed in core, and the Databricks provider does not need a change. One correction to the version boundary, though, because it matters a lot for anyone hitting this: the fix is already in 3.1.0, not only in >= 3.2.1. The affected window is the 3.0.x line.

Verified on a real workspace

I ran the same deferrable DatabricksRunNowOperator against a real Databricks job on each core version, with the provider pinned at 7.18.1 in every run (official apache/airflow images, airflow standalone, connection stored in the metadata DB so the triggerer has to fetch it through the Execution API):

core	result	trigger log
3.0.6	task failed on the first poll	RuntimeError: You cannot use AsyncToSync in the same thread as an async event loop, frame-for-frame identical to your traceback (triggerer_job_runner.py send -> asgiref/sync.py), then Trigger exited without sending an event. Dependent tasks will be failed.
3.1.0	task success	polled life_cycle_state: RUNNING twice, Trigger fired event, trigger completed
3.1.1	task success	same
3.2.2	task success	same

Since the provider was identical across all four runs, the difference is entirely core-side — which also confirms your point that no provider upgrade can help someone on 3.0.x.

Why 3.1.0 already works

The fix from #55799 landed in task-sdk's Connection.get() (not in the secrets backend, which is what makes it easy to miss when reading the diff):

except RuntimeError as e:
    if str(e).startswith("You cannot use AsyncToSync in the same thread as an async event loop"):
        import greenback
        task = asyncio.current_task()
        if greenback.has_portal(task):
            warnings.warn("You should not use sync calls here -- use `await Conn.async_get` instead", stacklevel=2)
            return greenback.await_(cls.async_get(conn_id))

That is exactly the path the Databricks hook takes: the 3.1.0 and 3.1.1 trigger logs contain that You should not use sync calls here warning, emitted from BaseHook.get_connection, immediately before the successful poll. So BaseDatabricksHook.databricks_conn staying a synchronous cached_property is fine from 3.1.0 onward — core absorbs it.

For completeness, the mechanism moved twice after that, which is why the code looks different depending on which version you read:

• 3.1.1 - 3.2.1: the same greenback fallback also present in ExecutionAPISecretsBackend.get_connection (the Fix connection access in triggerer for deferrable operators #57154 line of work).
• 3.2.2 and later, including main: the bridge moved one layer down into TriggerCommsDecoder.send() (Fix triggerer crash when multiple triggers call sync SDK methods concurrently #66412), so it now covers every sync SDK call made from a trigger's event-loop thread, not just connection fetches.

Suggested resolution

• This can be closed as fixed in core, with Airflow >= 3.1.0 as the requirement for deferrable Databricks operators. On MWAA 3.0.6 the only fix is a core upgrade; there is no provider-side workaround.
• Implement async version of databricks_conn in BaseDatabricksHook #55568 and fix(databricks): avoid AsyncToSync error in triggerer for deferrable operators #63775 were correctly closed — a Databricks-specific async databricks_conn is not needed, and adding one now would duplicate what core does for every provider.
• The one gap I do see is discoverability: the provider declares apache-airflow>=2.11.0, so it installs happily on 3.0.x where deferrable mode cannot work, and the resulting failure (AsyncToSync deep in asgiref) gives the user nothing to act on. A one-line note in the provider's deferrable docs stating that deferrable operators need core >= 3.1.0 would have answered this issue before it was filed. Happy to raise that as a small docs PR if maintainers think it is worth having.

---

Drafted-by: Claude Code (Opus 5); reviewed by @moomindani before posting

[jroachgolf84]

I think the takeaway here is that we want to do this.

The one gap I do see is discoverability: the provider declares apache-airflow>=2.11.0, so it installs happily on 3.0.x where deferrable mode cannot work, and the resulting failure (AsyncToSync deep in asgiref) gives the user nothing to act on. A one-line note in the provider's deferrable docs stating that deferrable operators need core >= 3.1.0 would have answered this issue before it was filed. Happy to raise that as a small docs PR if maintainers think it is worth having.

[LeonxLJX]

Hi! I'd like to work on this issue. Please let me know if I can be assigned. Thanks!