Bug: DatabricksExecutionTrigger raises RuntimeError: AsyncToSync in the same thread as an async event loop in Airflow 3

Apache Airflow Provider(s)

databricks

Versions of Apache Airflow Providers

7.9.1

Apache Airflow version

3.0.x

Operating System

macOS Tahoe 26.3.1

Deployment

Amazon (AWS) MWAA

Deployment details

We are in a current migration from MWAA Airflow 2.10.3 to Airflow 3.0.6. The deployment is being done using MWAA template, the only special thing to mention is the modified constraint file and the requirements.txt added here

constraints-dev.txt
requirements-dev.txt

What happened

When running a DatabricksTaskOperator (or any operator that creates a DatabricksExecutionTrigger) with deferrable=True in Airflow 3, the trigger fails immediately with:

  RuntimeError: You cannot use AsyncToSync in the same thread as an async event loop
  - just await the async function directly.

The entire stacktrace can be seen below:

[2026-03-16, 17:35:47] ERROR - Trigger failed:
Traceback (most recent call last):

  File "/usr/local/airflow/.local/lib/python3.12/site-packages/airflow/jobs/triggerer_job_runner.py", line 963, in cleanup_finished_triggers
    result = details["task"].result()
             ^^^^^^^^^^^^^^^^^^^^^^^^

  File "/usr/local/airflow/.local/lib/python3.12/site-packages/airflow/jobs/triggerer_job_runner.py", line 1072, in run_trigger
    async for event in trigger.run():

  File "/usr/local/airflow/.local/lib/python3.12/site-packages/airflow/providers/databricks/triggers/databricks.py", line 90, in run
    run_state = await self.hook.a_get_run_state(self.run_id)
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

  File "/usr/local/airflow/.local/lib/python3.12/site-packages/airflow/providers/databricks/hooks/databricks.py", line 533, in a_get_run_state
    response = await self._a_do_api_call(GET_RUN_ENDPOINT, json)
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

  File "/usr/local/airflow/.local/lib/python3.12/site-packages/airflow/providers/databricks/hooks/databricks_base.py", line 744, in _a_do_api_call
    url = self._endpoint_url(full_endpoint)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

  File "/usr/local/airflow/.local/lib/python3.12/site-packages/airflow/providers/databricks/hooks/databricks_base.py", line 654, in _endpoint_url
    port = f":{self.databricks_conn.port}" if self.databricks_conn.port else ""
                                              ^^^^^^^^^^^^^^^^^^^^

  File "/usr/local/lib/python3.12/functools.py", line 998, in __get__
    val = self.func(instance)
          ^^^^^^^^^^^^^^^^^^^

  File "/usr/local/airflow/.local/lib/python3.12/site-packages/airflow/providers/databricks/hooks/databricks_base.py", line 145, in databricks_conn
    return self.get_connection(self.databricks_conn_id)  # type: ignore[return-value]
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

  File "/usr/local/airflow/.local/lib/python3.12/site-packages/airflow/hooks/base.py", line 64, in get_connection
    conn = Connection.get_connection_from_secrets(conn_id)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

  File "/usr/local/airflow/.local/lib/python3.12/site-packages/airflow/models/connection.py", line 478, in get_connection_from_secrets
    conn = TaskSDKConnection.get(conn_id=conn_id)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

  File "/usr/local/airflow/.local/lib/python3.12/site-packages/airflow/sdk/definitions/connection.py", line 144, in get
    return _get_connection(conn_id)
           ^^^^^^^^^^^^^^^^^^^^^^^^

  File "/usr/local/airflow/.local/lib/python3.12/site-packages/airflow/sdk/execution_time/context.py", line 160, in _get_connection
    msg = SUPERVISOR_COMMS.send(GetConnection(conn_id=conn_id))
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

  File "/usr/local/airflow/.local/lib/python3.12/site-packages/airflow/jobs/triggerer_job_runner.py", line 740, in send
    return async_to_sync(self.asend)(msg)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

  File "/usr/local/airflow/.local/lib/python3.12/site-packages/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.

What you think should happen instead

No response

How to reproduce

Prerequisites: Airflow 3.0.x with a working Databricks connection (databricks_default).

  # dags/test_deferrable_databricks.py
  from datetime import datetime, timezone
  from airflow import DAG
  from airflow.providers.databricks.operators.databricks import DatabricksTaskOperator

  with DAG(
      dag_id="test_deferrable_databricks",
      start_date=datetime(2024, 1, 1, tzinfo=timezone.utc),
      schedule=None,
      catchup=False,
  ):
      DatabricksTaskOperator(
          task_id="test_task",
          databricks_conn_id="databricks_default",
          deferrable=True,  # <-- triggers the bug
          task_config={
              "spark_python_task": {
                  "python_file": "dbfs:/some/script.py",
              },
          },
          new_cluster={
              "spark_version": "13.3.x-scala2.12",
              "node_type_id": "i3.xlarge",
              "num_workers": 1,
          },
      )

Anything else

No response

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

I have analyzed this issue and it's a valid regression in Airflow 3. The DatabricksExecutionTrigger crashes because it accesses synchronous cached_propertys (like databricks_conn) from within an async event loop. This triggers a sync connection lookup in the Task SDK that uses async_to_sync, leading to the reported RuntimeError.
The fix is straightforward: we can asynchronously pre-populate the connection cache in BaseDatabricksHook.aenter before the polling loop starts. I have a solution ready and would be happy to submit a PR with this fix and accompanying tests.

I don't believe is a simple warmup of cached hooks that the operator needs. I tried that but the trigger gets stuck after pausing

[2026-03-17, 14:36:32] INFO - Current state of the databricks task c3_dynamo_replication_hourly__tasks__dynamo_master is PENDING: source="custom operator that inherits from the DatabricksTaskOperator and applies a mixin that fixes the trigger"
[2026-03-17, 14:36:32] INFO - Connection Retrieved 'databricks_default': source="airflow.hooks.base"
[2026-03-17, 14:36:32] INFO - Pausing task as DEFERRED. : dag_id="c3_dynamo_replication_hourly": task_id="tasks.dynamo_master": run_id="manual__2026-03-17T17:36:18.160557+00:00": source="task"

It wont recover after that, the trigger doesn't handle well the async calls

EDIT:

The problem that caused the operator to get stuck was an error of when I warmed up the cached properties. I created a class that inherits from DatabricksExecutionTrigger and its __init__() did the warmup (didn't work). I was able to fix it by overriding the run method of this trigger, making a warmup previous to the super.run() method

    async def run(self):
        # Awaiting SUPERVISOR_COMMS.asend
        # - It stays on the triggerer's event loop (no async_to_sync)
        # - It uses the loop's own StreamReader (no "different loop" error)
        # Once hook.__dict__ is populated, cached_property never calls get_connection.
        from airflow.sdk.execution_time.comms import ErrorResponse, GetConnection
        from airflow.sdk.execution_time.context import _convert_connection_result_conn
        from airflow.sdk.execution_time.task_runner import SUPERVISOR_COMMS

        response = await SUPERVISOR_COMMS.asend(GetConnection(conn_id=self.hook.databricks_conn_id))
        if isinstance(response, ErrorResponse):
            raise RuntimeError(f"Failed to fetch Databricks connection '{self.hook.databricks_conn_id}': {response}")

        self.hook.__dict__["databricks_conn"] = _convert_connection_result_conn(response)
        _ = self.hook.host  # pylint: disable=pointless-statement  # safe now that databricks_conn is cached

        async for event in super().run():
            yield event

@mwojtyczka

cc @mwojtyczka @alexott @moomindani maybe one of you can take a look?

Depending on how you solve this bug, you may face a new bug where the triggerer will not take the latest task_run_id of the workflow.

If the task fails in databricks, and retries successfully later, the airlfow operator may not see that success state. The solution I proposed above had this issue

@eladkal

Thanks for the ping @eladkal. I've been looking at this from the Databricks perspective.

The bug report looks valid — the stack trace clearly shows the root cause.

One thing worth noting: this isn't specific to the Databricks provider. The root cause is that a sync @cached_property calls get_connection() inside the triggerer's async event loop, hitting async_to_sync(). I checked other providers and found at least the dbt Cloud provider has the same pattern (DbtCloudHook.connection → get_connection() used from DbtCloudRunJobTrigger). Some providers (e.g. Kubernetes) avoid this by using get_async_connection(). A framework-level guideline or guard might be worth considering in addition to the per-provider fix.

@moomindani

@moomindani hopefully #66412 solves the issue from the root.

Navigation Menu

Repository navigation

Bug: DatabricksExecutionTrigger raises RuntimeError: AsyncToSync in the same thread as an async event loop in Airflow 3 #63760

Apache Airflow Provider(s)

Versions of Apache Airflow Providers

Apache Airflow version

Operating System

Deployment

Deployment details

What happened

What you think should happen instead

How to reproduce

Anything else

Are you willing to submit PR?

Code of Conduct

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Bug: DatabricksExecutionTrigger raises RuntimeError: AsyncToSync in the same thread as an async event loop in Airflow 3 #63760

Description

Apache Airflow Provider(s)

Versions of Apache Airflow Providers

Apache Airflow version

Operating System

Deployment

Deployment details

What happened

What you think should happen instead

How to reproduce

Anything else

Are you willing to submit PR?

Code of Conduct

Activity

Subham-KRLX commented on Mar 17, 2026

fedemgp commented on Mar 17, 2026

eladkal commented on Apr 9, 2026

fedemgp commented on Apr 14, 2026

moomindani commented on Apr 15, 2026

eladkal commented on May 5, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions