Avoid extra_dejson in ADF and Synapse async hooks by Vamsi-klu · Pull Request #72130 · apache/airflow

Vamsi-klu · 2026-08-27T06:54:22Z

What is the change?

The async paths in the Azure Data Factory and Synapse pipeline hooks stop reading conn.extra_dejson and parse extras with json.loads(conn.extra) instead. AzureSynapsePipelineAsyncHook.get_async_conn now fetches the connection through get_async_connection rather than the sync get_connection, and ADF refresh_conn closes the cached _async_conn before rebuilding it.

Why did I do it?

extra_dejson can call mask_secret, which does a sync send on the triggerer event loop and raises RuntimeError: You cannot use AsyncToSync in the same thread as an async event loop. The stack users see at execute_complete is that trigger error re-raised. This is the same leftover that #55179 fixed for MSGraph.

How did I do it?

In hooks/data_factory.py and hooks/synapse.py I swapped conn.extra_dejson for json.loads(conn.extra) if conn.extra else {} on the async paths only, and made refresh_conn await self.close(). The sync hooks keep extra_dejson so worker-side secret masking is unchanged.

What's the impact?

Deferrable ADF and Synapse pipeline runs stop crashing on the triggerer with the AsyncToSync error. Sync hooks and execute_complete behave exactly as before. Azure provider only.

What's the test plan?

New unit tests make extra_dejson raise the real AsyncToSync error and assert the async hooks still build their clients; they fail if extra_dejson comes back. test_refresh_conn now asserts the aio client gets closed. Reviewers can re-run the two touched test files or breeze testing providers-tests --test-type "Providers[microsoft.azure]"; provider CI covers the same.

Was generative AI tooling used to co-author this PR?

Yes

Generated-by: Cursor Grok 4.6 following the guidelines

extra_dejson can mask secrets via a sync send on the triggerer event loop, which raises AsyncToSync. Parse extras with json.loads instead, matching the MSGraph workaround. closes: apache#55728

dabla · 2026-08-28T06:58:12Z

Could you review the leftover extra_dejson and sync get_connection hops on the ADF and Synapse async hooks? This is the same AsyncToSync triggerer-loop failure as #55179, now in AzureDataFactoryAsyncHook.get_async_conn / provide_targeted_factory_async and AzureSynapsePipelineAsyncHook.get_async_conn. The question is whether json.loads(conn.extra) on the async path, and get_async_connection in the Synapse hook, is the right leftover fix, and whether leaving the sync hooks on extra_dejson is correct.

cc @dabla @eladkal @Lee-W

Drafted-by: Cursor Grok 4.6 (no human review before posting)

There is no async extra_dejson counterpart, so yes I think it makes senses. We could add one but then we also need to add a backward compat function for the providers, which probably would be the neatest solution.

The is also a PR open which addresses another generic issue regarding the async path with connections as well.

potiuk

Nice. Simple to accept without proof of running it on real Synapse

potiuk · 2026-08-29T13:24:20Z

But @Vamsi-klu - please test it when we release the provider - you will be `@-mentioned'

extra_dejson can mask secrets via a sync send on the triggerer event loop, which raises AsyncToSync. Parse extras with json.loads instead, matching the MSGraph workaround. closes: apache#55728

get_conn() resolves the connection with get_connection() and reads extra_dejson, whose secret masking sends to the supervisor synchronously; inside an async task with another async SDK call in flight both raise DeadlockImminentError. aget_conn() returns an AsyncOpenAI client configured exactly like get_conn()'s, from get_async_connection() with the extra from Connection.aextra_dejson() (apache#71890, Airflow 3.3.2+), or the raw extra via json.loads on older versions (as in apache#72130). The client settings move into _client_kwargs(), shared by both paths. acreate_embeddings() is the async counterpart of create_embeddings(). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

get_conn() resolves the hook's connection and every fallback connection with get_connection() and reads extra_dejson, whose secret masking sends to the supervisor synchronously; inside an async task with another async SDK call in flight both raise DeadlockImminentError, although the agent the hook builds is async-native. aget_conn() fetches the connection and its fallbacks with get_async_connection() (extra from Connection.aextra_dejson(), apache#71890, Airflow 3.3.2+, or the raw extra via json.loads on older versions as in apache#72130), seeds the existing connection cache and builds the model with the unchanged get_conn(); acreate_agent() does the same before create_agent(), spec_file included. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

get_conn() resolves the connection with get_connection() and reads extra_dejson, whose secret masking sends to the supervisor synchronously; inside an async task with another async SDK call in flight both raise DeadlockImminentError. aget_conn() returns an AsyncOpenAI client configured exactly like get_conn()'s, from get_async_connection() with the extra from Connection.aextra_dejson() (apache#71890, Airflow 3.3.2+), or the raw extra via json.loads on older versions (as in apache#72130). The client settings move into _client_kwargs(), shared by both paths. acreate_embeddings() is the async counterpart of create_embeddings(). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

get_conn() resolves the hook's connection and every fallback connection with get_connection() and reads extra_dejson, whose secret masking sends to the supervisor synchronously; inside an async task with another async SDK call in flight both raise DeadlockImminentError, although the agent the hook builds is async-native. aget_conn() fetches the connection and its fallbacks with get_async_connection() (extra from Connection.aextra_dejson(), apache#71890, Airflow 3.3.2+, or the raw extra via json.loads on older versions as in apache#72130), seeds the existing connection cache and builds the model with the unchanged get_conn(); acreate_agent() does the same before create_agent(), spec_file included. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

…74147) Connection.extra_dejson masks the extra's secrets with a synchronous call to the supervisor, which raises DeadlockImminentError when a hook reads it on an event loop with another async call in flight. Airflow 3.3.2+ has Connection.aextra_dejson() (#71890), but providers that still support older versions cannot call it directly, so async hooks fall back to json.loads(conn.extra) and skip the masking (#72130). get_async_extra_dejson(conn) awaits Connection.aextra_dejson() when it exists, and otherwise runs extra_dejson in a worker thread, where blocking on the supervisor is safe, as get_async_connection() does for get_connection(). Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

* Avoid the worker-thread fallback of get_async_extra_dejson on Airflow 3.0 and 3.1 get_async_extra_dejson() falls back to running extra_dejson in a worker thread when Connection.aextra_dejson() is missing. Its secret masking sends to the supervisor, and on Airflow 3.0 and 3.1 the supervisor channel has no thread lock: a send from a worker thread could interleave with an asend() in flight on the event loop, e.g. in the triggerer. On those versions the helper now deserializes Connection.extra without the supervisor, as async hooks did before (#55179, #72130). The worker thread stays for Airflow 3.2 to 3.3.1, whose channel is thread-safe, and Airflow 2, which has no supervisor. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * Test get_async_extra_dejson against the real supervisor comms instead Revert the Airflow 3.0/3.1 json.loads branch: on those versions async code only runs in the triggerer, whose channel serializes every request, a worker thread's synchronous send included, through asend() and an asyncio.Lock, so the worker-thread fallback was safe and kept the masking. Instead, add a test that forks a real task process talking to the real supervisor of the installed Airflow and reads extras concurrently from an event loop, so each compat job checks the helper against that version's comms rather than mocks. It needs Airflow 3.2+, the first version where a task process can call the supervisor from async code. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * Import AIRFLOW_V_3_2_PLUS from tests_common in the supervisor comms test Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

Avoid extra_dejson in ADF and Synapse async hooks

ab60f51

extra_dejson can mask secrets via a sync send on the triggerer event loop, which raises AsyncToSync. Parse extras with json.loads instead, matching the MSGraph workaround. closes: apache#55728

boring-cyborg Bot added area:providers provider:microsoft-azure Azure-related issues labels Aug 27, 2026

This comment was marked as spam.

Sign in to view

potiuk approved these changes Aug 29, 2026

View reviewed changes

potiuk merged commit 39b4804 into apache:main Aug 29, 2026
83 checks passed

vincbeck mentioned this pull request Sep 10, 2026

Status of testing Providers that were prepared on September 09, 2026 #72902

Closed

dabla mentioned this pull request Oct 3, 2026

Add get_async_extra_dejson to common.compat connection helpers #74147

Merged

1 task done

This was referenced Oct 3, 2026

Add async aget_conn and acreate_embeddings to OpenAIHook #74149

Open

Add async aget_conn and acreate_agent to PydanticAIHook #74150

Open

dabla mentioned this pull request Oct 3, 2026

Test get_async_extra_dejson against the real supervisor comms #74166

Merged

1 task done

Navigation Menu

Repository navigation

Avoid extra_dejson in ADF and Synapse async hooks - #72130

Avoid extra_dejson in ADF and Synapse async hooks#72130
potiuk merged 1 commit into
apache:mainfrom
Vamsi-klu:fix/55728-adf-async-extra-dejson

Vamsi-klu commented Aug 27, 2026 •

edited

Loading

Uh oh!

This comment was marked as spam.

dabla commented Aug 28, 2026

Uh oh!

potiuk left a comment

Uh oh!

potiuk commented Aug 29, 2026

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Conversation

Vamsi-klu commented Aug 27, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What is the change?

Why did I do it?

How did I do it?

What's the impact?

What's the test plan?

Was generative AI tooling used to co-author this PR?

Uh oh!

This comment was marked as spam.

dabla commented Aug 28, 2026

Uh oh!

potiuk left a comment

Choose a reason for hiding this comment

Uh oh!

potiuk commented Aug 29, 2026

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Vamsi-klu commented Aug 27, 2026 •

edited

Loading