[sunank200]

This PR donates the following HttpHookAsync that interacts with HTTP servers using Python Async. This was developed in astronomer-providers repo to apache airflow.

cc: @kaxil

^ Add meaningful description above

Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

[Taragolis]

Is it intend to get connection for every run call?

I see that the same behaviour exists in Requests-based hook but I don't think this is a good approach, most of the time we would spend for get connection.

[Taragolis]

And also don't know what happen when multiple different Trigger's will try to call sync_to_async simultaneously?

• Is is thread safety?
• Or is all of them would run sequentially?

[sunank200]

@Taragolis when we call sync_to_async simultaneously these are concurrent calls. This is not multithreading.sync_to_async lets async code call a synchronous function, which is run in a threadpool and control returned to the async coroutine when the synchronous function completes. So it concurrency than multithreading

[Taragolis]

Well, I found a time for check this magic. As I could see sync_to_async run in ThreadPoolExecutor which is use internally threads. However for avoid any thread related issues it run in pool with single worker if thread_sensitive=True.

https://github.com/django/asgiref/blob/730d865506adf565d53e5a67cceb9f796dc472ba/asgiref/sync.py#L345-L346

It will run all functions/methods decorated by sync_to_async sequentially one after another as result no concurrency here, and one function which hung, for example due to infinity timeout to some service which do not send response, would block others

That is some simple sample

import asyncio
import datetime
import time
from asgiref.sync import sync_to_async


@sync_to_async
def task(value):
    print(f"{datetime.datetime.utcnow().isoformat()}, sync_to_async start {value!r}")
    time.sleep(3)
    print(f"{datetime.datetime.utcnow().isoformat()}, sync_to_async end {value!r}")

async def async_task(value):
    print(f"{datetime.datetime.utcnow().isoformat()}, async_task start {value!r}")
    await asyncio.sleep(3)
    print(f"{datetime.datetime.utcnow().isoformat()}, async_task end {value!r}")

async def main():
    await asyncio.gather(
        asyncio.create_task(task(1)),
        asyncio.create_task(task(2)),
        asyncio.create_task(async_task(3)),
        asyncio.create_task(async_task(4)),
        asyncio.create_task(task(5)),
        asyncio.create_task(async_task(6)),
    )

if __name__ == "__main__":
    asyncio.run(main())

[sunank200]

@Taragolis we have already used sync_to_async in various airflow provider's code base currently as well like: https://github.com/apache/airflow/blob/main/airflow/providers/google/common/hooks/base_google.py#L629

@andrewgodwin @kaxil WDYT?

[kaxil]

More details here: https://docs.djangoproject.com/en/4.1/topics/async/#sync-to-async

[Taragolis]

The main point that sync_to_async decorated functions run in some kind of sequential queue (or otherwise it not thread safe), as result It may lead some issues with thread safety.

Code which run inside sync_to_async not lock native async code but lock execution other sync_to_async decorated code it is demonstrated in my example. The same happen in triggerer airflow service when single hanged sync_to_async decorated function might cause that one of triggered would have huge queue for that code, an almost every single Trigger call sync_to_async.

So, unfortunetly ,"we have already used sync_to_async in various airflow provider's code base currently as well like" not make things better rather than create some kind of time bomb.

I don't mean that you doing a bad job and all this sync_to_async decorated functions added for a good purpose, and in general it nice to have more deferrable tasks out-of-the-box however the main problem that Airflow core not ready to operate with Deferrable tasks which need to have access to Core components, such as Connections, Variables, XCom and etc.

[Taragolis]

Maybe better to have exponential backoff retry?

[sunank200]

Could you elaborate why exponential backoff retry would be better suited here?

[Taragolis]

For unknown pattern of usage exponential backoff usual is a better:

• It add throttling for requests
• If service temporary down more chance that with exponential backoff request finish successfully in the end.

[sungwy]

@kaxil @sunank200 I'm not an expert on this provider, but it looks like test_do_api_call_async_retryable_error at tests/providers/http/hooks/test_http.py:436 was committed with a failing test. It seems to me that the underlying aioreponses request_mock call is raising a ClientConnectionError, which isn't caught in HttpAsyncHook did you have a specific way you wanted to handle the ClientConnectionError? If not, I could try to treat it similarly to how you are handling ClientResponseError and patch it up on my open PR: b6ba1cc

[sunank200]

it looks like test_do_api_call_async_retryable_error at tests/providers/http/hooks/test_http.py:436 was committed with a failing test. It seems to me that the underlying aioreponses request_mock call is raising a ClientConnectionError, which isn't caught in HttpAsyncHook

@syun64 I think the test missed the pytest fixture aioresponse. Added it in this PR: #29540 Please review.