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>

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>

* Add async aget_conn and acreate_embeddings to OpenAIHook 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() (#71890, Airflow 3.3.2+), or the raw extra via json.loads on older versions (as in #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> * Use get_async_extra_dejson from common.compat Replace the local aextra_dejson fallback with the common.compat helper added in #74147, which masks the extra's secrets asynchronously on Airflow 3.3.2+ and reads extra_dejson from a worker thread on older versions. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * Spec the mocked AsyncOpenAI client in test_acreate_embeddings MagicMock() without a spec accepts any attribute, so a typo on the async path such as client.embedding would still pass. spec=AsyncOpenAI guards the client's namespace the way the sync tests' spec=OpenAI does. * Assert aget_conn reads the extra through aextra_dejson The connection mock carried an extra string and no test checked that aextra_dejson() was awaited, so a hook reading json.loads(conn.extra), which skips the secret masking, kept every test green. Drop extra from the mock's spec and assert the await in the tests that have it. * Share the embeddings result shaping between the sync and async paths acreate_embeddings copied the three lines that pick one vector for a single input and sort one vector per item for a batch. Move them into _embeddings_result, used by create_embeddings and acreate_embeddings, so the two paths cannot drift, like _client_kwargs for the client. * Resolve the async connection through the hook get_conn resolves through self.get_connection, so a subclass override applies there, but aget_conn called get_async_connection without hook= and the lookup went through BaseHook. Pass hook=self, as AnthropicHook.get_async_conn does, and test it with a subclass that overrides aget_connection, through the real common.compat helper. * Build the async client per call instead of caching it on the hook _aget_cached_conn checked the cache before two awaits, so concurrent calls on a fresh hook each built a client, dropped all but one unclosed, and the kept one was never closed and stayed bound to the first event loop. aget_conn now builds a client per call and says to close it, as AnthropicHook.get_async_conn does; acreate_embeddings closes its client with async with. Tested with three concurrent calls on one hook. * State that aget_conn avoids synchronous Task SDK calls from Airflow 3.3.2 Before 3.3.2, core's aget_connection still masks the connection's secrets synchronously, so the docstring promised more than older cores give. * Pin the per-call client in the concurrent embeddings test Patching aget_conn let the test supply distinct clients itself, so it kept passing if aget_conn cached one client on the hook again. * Note the token exchange per client for workload identity in aget_conn The SDK caches the exchanged token on each client, so callers making many calls should know they can keep one client open and reuse it. * Require common.compat 1.21.0 for get_async_extra_dejson The hook imports get_async_extra_dejson at module level, so with an older common.compat the whole hook, sync get_conn included, fails to import. * Restore the use-next-version marker on the common.compat dependency Only release managers raise provider dependency floors; the marker lets them bump it to the release that ships get_async_extra_dejson. * Build the embeddings usage as a Usage model in the async test client mypy checks the annotated _async_client helper and rejects the dict passed to CreateEmbeddingResponse.usage, which is typed Usage. * Update providers/openai/src/airflow/providers/openai/hooks/openai.py Co-authored-by: Kaxil Naik <kaxilnaik@gmail.com> * Update providers/openai/src/airflow/providers/openai/hooks/openai.py Co-authored-by: Kaxil Naik <kaxilnaik@gmail.com> * Update providers/openai/src/airflow/providers/openai/hooks/openai.py Co-authored-by: Kaxil Naik <kaxilnaik@gmail.com> * Update providers/openai/tests/unit/openai/hooks/test_openai.py Co-authored-by: Kaxil Naik <kaxilnaik@gmail.com> * Drop test docstrings that repeat the test names --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: Kaxil Naik <kaxilnaik@gmail.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>

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

Merged

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

potiuk mentioned this pull request Oct 8, 2026

Status of testing Providers that were prepared on October 06, 2026 #74459

Closed

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

Uh oh!

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

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Vamsi-klu commented Aug 27, 2026 •

edited

Loading