Connection Timeout between prefect & Airbyte

[lucienfregosibodyguard]

Hi

For some flows / airbyte tasks I got some timeout when prefect is trying to get the airbyte task job status

Encountered exception during execution:
Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/httpcore/backends/asyncio.py", line 33, in read
    return await self._stream.receive(max_bytes=max_bytes)
  File "/usr/local/lib/python3.9/site-packages/anyio/_backends/_asyncio.py", line 1265, in receive
    await self._protocol.read_event.wait()
  File "/usr/local/lib/python3.9/asyncio/locks.py", line 226, in wait
    await fut
asyncio.exceptions.CancelledError

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/httpcore/_exceptions.py", line 8, in map_exceptions
    yield
  File "/usr/local/lib/python3.9/site-packages/httpcore/backends/asyncio.py", line 35, in read
    return b""
  File "/usr/local/lib/python3.9/site-packages/anyio/_core/_tasks.py", line 118, in __exit__
    raise TimeoutError
TimeoutError

The weird thing is that happens quite randomly, it happens on some task but some others are working well.

Any idea ?
Maybe at least we could have a retry on the request and not raise an exception directly ?

[zzstoatzz]

Hi @lucienfregosibodyguard!

Could you please provide an example showing how you were using a prefect-airbyte task?

If possible, it would also be helpful to see the part of the stack trace involving the calls within that task - from the portion provided, it's unclear how the error relates to the implementation.

To handle potential transient failures, you could use the with_options method on Task to add retries to trigger_sync after you import it, like:

from prefect import flow
from prefect_airbyte.connections import trigger_sync

@flow
def airbyte_orchestrator(connection_id: str = 'xxx-xxx-xxx'):
   trigger_sync.with_options(retries=3, retry_delay_seconds=5)(connection_id=connection_id)

[nyuta01]

Similar problems are occurring.
I am wondering if this problem could be caused by the default timeout of 5 seconds for httpx.AsyncClient used within AirbyteClient?
If this is correct, it would be greatly appreciated if you could add a timeout parameter to trigger_sync.

https://github.com/PrefectHQ/prefect-airbyte/blob/v0.1.0/prefect_airbyte/client.py#L56
https://github.com/encode/httpx/blob/0.23.0/httpx/_client.py#L1355
https://github.com/encode/httpx/blob/0.23.0/httpx/_config.py#L361

[lucienfregosibodyguard]

Hi @zzstoatzz

Here are the code where I'm using airbyte task

        res = trigger_sync(
            connection_id=connection_id,
            poll_interval_s=3,
            status_updates=True,
            airbyte_server_host=airbyte_server_host,
            airbyte_server_port=airbyte_server_port,
        )

And the full stack trace :
https://pastebin.com/zuDYgpGe

I didn't know for the withoptions I will try with this.

[zzstoatzz]

hi @nyuta01 and @lucienfregosibodyguard !

I went ahead and added a timeout parameter to the trigger_sync task (as well as the export_configuration task) which will set the timeout for requests made by the httpx.AsyncClient.

prefect-airbyte==0.1.1 includes this change and has just been released - thank you both for raising this issue!