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

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

Airflow 3.0.6, apache-airflow-providers-databricks==7.7.1 (also reproduces on 7.18.1's unchanged code path per the above).
Any DAG using DatabricksRunNowOperator(deferrable=True, job_name=..., databricks_conn_id=...).
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

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.

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.

@moomindani

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

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.

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

@moomindani

Quick context before you spend time on this, since the thread is long: the underlying crash is already fixed in Airflow core from 3.1.0 onwards, and the affected window is the 3.0.x line. I verified that boundary with a real-environment matrix (provider pinned at 7.18.1 in every run, so core was the only variable): 3.0.6 reproduces the traceback frame-for-frame, while 3.1.0, 3.1.1 and 3.2.2 all poll and fire the event normally — details in #71525 (comment).

There are also two open PRs pointing at this issue already:

Fix AsyncToSync error in Databricks deferrable operators by fetching connection asynchronously #71840 — provider-side fix, fetches the connection asynchronously so deferrable mode also works on 3.0.x.
Document Airflow 3 requirement for Databricks deferral #71667 — documents the core version floor for Databricks deferral.

So a third implementation would most likely duplicate one of those. If you would like to help this along, reviewing #71840 (it has had no review since it was opened) is probably worth more than a new patch. Assignment is a committer's call, not mine.

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

Hey, I am new to Open Source and would like to give this issue a try if still available.

I'd like to work on this. Looks like the AsyncToSync call in the Databricks trigger/hook path is getting invoked from within an already-running event loop in the triggerer process, which is what's blowing up. I'll dig into the deferrable code path and put together a fix plus a regression test. Will open a PR shortly.

Hi! I'd like to fix DatabricksRunNowOperator(deferrable=True) crashing in the triggerer with AsyncTask. Plan: reproduce in the triggerer, trace the async task path, fix and add a regression test. May I be assigned?

@moomindani

Three more offers to take this since my note above, so worth restating once for anyone landing here: the crash is already fixed in Airflow core from 3.1.0 onwards and only the 3.0.x line is affected — verified with a real-environment matrix, provider pinned in every run (details in #71525 (comment)).

Two open PRs already point at this issue: #71840 (provider-side fix, so deferrable mode works on 3.0.x too) and #71667 (documents the core version floor). A fresh patch would duplicate one of them. #71840 has had no review from anyone but me, so reviewing it moves this along faster than a third implementation would.

The good first issue label is what is drawing these requests and it no longer fits — the work is claimed twice over. Worth a committer removing it, or closing this issue as covered once one of the two PRs lands.

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

@moomindani

Four PRs now point at this issue — #71840, #72014, #72077, #72245 — and only one has had a review, so here is the comparison in one place. I am not a committer; this is the evidence, not a decision.

The constraint that decides most of it: the crash only happens on core 3.0.x (version matrix), and the provider supports apache-airflow>=2.11.0. BaseHook.aget_connection first ships in 3.1.0 (added in b83fcf9a31; the earliest release tag containing it is 3.1.0). A fix built on aget_connection alone therefore only runs on the versions where the bug is already gone.

PR	how it gets the connection	works on 3.0.x	tests	CI
#71840	`get_async_connection(..., hook=self)` from common.compat — native `aget_connection` on 3.1+, `sync_to_async` fallback below	yes	yes	red, see below
#72245	the same compat helper	yes	none in the diff	not run (only Mergeable/WIP)
#72077	`sync_to_async(self.get_connection)` directly	yes	yes	not run (only Mergeable/WIP)
#72014	`BaseHook.aget_connection`	no	yes, but in `test_databricks.py` while the change is in `databricks_base.py`	red

#71840 is the closest to mergeable, and two small things stand between it and green CI:

get_async_connection is called at databricks_base.py:201 but never imported. That single omission is the whole of the ruff and mypy failure (Name "get_async_connection" is not defined), and every downstream job fails on the same NameError rather than on anything about the approach.
the hook= keyword arrived with common.compat 1.17.0 (Honor async hook subclass overrides in get_async_connection #69140), while the provider pins apache-airflow-providers-common-compat>=1.13.0. That bound needs raising — or drop hook= and lose the subclass override it exists to honour.

#72077's approach also works, and it carries no compat dependency, but it hardcodes the sync_to_async fallback and so never takes the native async path on 3.1+ — which is the choice the compat helper exists to make.

#72245 additionally adds a_get_run_tasks, a_cancel_run and a_cancel_sql_statement, which are unrelated to this crash and would be easier to review as their own PR.

Whichever lands, #71667 documents the version floor and stays useful alongside it.

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

Fix AsyncToSync runtime error in Databricks deferrable operators

Fix AsyncToSync error in Databricks deferrable operators on Airflow 3.0

Fix AsyncToSync error in Databricks deferrable operators on Airflow 3…

Navigation Menu

Repository navigation

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

Apache Airflow version

If "Other Airflow 2 version" selected, which one?

What happened?

What you think should happen instead?

How to reproduce

Operating System

Versions of Apache Airflow Providers

Deployment

Deployment details

Anything else?

Are you willing to submit PR?

Code of Conduct

Verified on a real workspace

Why 3.1.0 already works

Suggested resolution

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

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

Description

Apache Airflow version

If "Other Airflow 2 version" selected, which one?

What happened?

What you think should happen instead?

How to reproduce

Operating System

Versions of Apache Airflow Providers

Deployment

Deployment details

Anything else?

Are you willing to submit PR?

Code of Conduct

Activity

boring-cyborg commented on Aug 13, 2026

anish-heidi commented on Aug 13, 2026

eladkal commented on Aug 14, 2026

moomindani commented on Aug 15, 2026

Verified on a real workspace

Why 3.1.0 already works

Suggested resolution

jroachgolf84 commented on Aug 15, 2026

LeonxLJX commented on Aug 27, 2026

moomindani commented on Aug 28, 2026

jasperjonkhans commented on Aug 29, 2026

avanish-garg commented on Sep 1, 2026

LeonxLJX commented on Sep 1, 2026

moomindani commented on Sep 3, 2026

moomindani commented on Sep 5, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions