https://prefect.io logo
Join Slack
Powered by
Hi <@ULVA73B9P> I’m using the `prefect-airbyte` l...
# ask-marvin
e
Hi @Marvin I’m using the 
prefect-airbyte
library to sync Airbyte Connections. Airbyte is running on Kubernetes (AKS). I’ve faced an error when Airbyte restarts it’s server or worker pods leading the Prefect flow failures. The airbyte sync continues to run. Below is the log. Can you advise how to debug further or fix it?
Copy code
Runner 'elt-deployment' submitting flow run '0177e1fb-a1a4-49e5-a2d7-62f5ef433ab7'
05:41:10 PM
prefect.flow_runs.runner
Opening process...
05:41:10 PM
prefect.flow_runs.runner
Completed submission of flow run '0177e1fb-a1a4-49e5-a2d7-62f5ef433ab7'
05:41:10 PM
prefect.flow_runs.runner
Downloading flow code from storage at '.'
05:41:12 PM
prefect.flow_runs
Created subflow run 'hospitable-goat' for flow 'airbyte_flow'
05:41:12 PM
prefect.flow_runs
Encountered exception during execution:
Traceback (most recent call last):
  File "/usr/local/lib/python3.11/site-packages/prefect_airbyte/client.py", line 220, in get_job_info
    response.raise_for_status()
  File "/usr/local/lib/python3.11/site-packages/httpx/_models.py", line 761, in raise_for_status
    raise HTTPStatusError(message, request=request, response=self)
httpx.HTTPStatusError: Server error '500 Internal Server Error' for url '<http://airbyte-airbyte-server-svc.airbyte.svc.cluster.local:8001/api/v1/jobs/get/>'
For more information check: <https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/500>

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/usr/local/lib/python3.11/site-packages/prefect/engine.py", line 863, in orchestrate_flow_run
    result = await flow_call.aresult()
             ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/_internal/concurrency/calls.py", line 326, in aresult
    return await asyncio.wrap_future(self.future)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/_internal/concurrency/calls.py", line 351, in _run_sync
    result = self.fn(*self.args, **self.kwargs)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/prefect/flows/elt_flow.py", line 7, in elt_flow
    airbyte_flow()
  File "/usr/local/lib/python3.11/site-packages/prefect/flows.py", line 1183, in __call__
    return enter_flow_run_engine_from_flow_call(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/engine.py", line 295, in enter_flow_run_engine_from_flow_call
    retval = from_sync.wait_for_call_in_loop_thread(
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/_internal/concurrency/api.py", line 243, in wait_for_call_in_loop_thread
    return call.result()
           ^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/_internal/concurrency/calls.py", line 317, in result
    return self.future.result(timeout=timeout)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/_internal/concurrency/calls.py", line 178, in result
    return self.__get_result()
           ^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/concurrent/futures/_base.py", line 401, in __get_result
    raise self._exception
  File "/usr/local/lib/python3.11/site-packages/prefect/_internal/concurrency/calls.py", line 388, in _run_async
    result = await coro
             ^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/client/utilities.py", line 51, in with_injected_client
    return await fn(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/engine.py", line 747, in create_and_begin_subflow_run
    return await terminal_state.result(fetch=True)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/states.py", line 91, in _get_state_result
    raise await get_state_exception(state)
  File "/usr/local/lib/python3.11/site-packages/prefect/engine.py", line 863, in orchestrate_flow_run
    result = await flow_call.aresult()
             ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/_internal/concurrency/calls.py", line 326, in aresult
    return await asyncio.wrap_future(self.future)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/_internal/concurrency/calls.py", line 388, in _run_async
    result = await coro
             ^^^^^^^^^^
  File "/opt/prefect/flows/airbyte_flow.py", line 115, in airbyte_flow
    await sync_chunked_connections(
  File "/opt/prefect/flows/airbyte_flow.py", line 79, in sync_chunked_connections
    await asyncio.gather(*parallel_subflows)
  File "/usr/local/lib/python3.11/site-packages/prefect/_internal/concurrency/api.py", line 182, in wait_for_call_in_loop_thread
    return call.result()
           ^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/_internal/concurrency/calls.py", line 317, in result
    return self.future.result(timeout=timeout)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/_internal/concurrency/calls.py", line 178, in result
    return self.__get_result()
           ^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/concurrent/futures/_base.py", line 401, in __get_result
    raise self._exception
  File "/usr/local/lib/python3.11/site-packages/prefect/_internal/concurrency/calls.py", line 388, in _run_async
    result = await coro
             ^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/client/utilities.py", line 51, in with_injected_client
    return await fn(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/engine.py", line 747, in create_and_begin_subflow_run
    return await terminal_state.result(fetch=True)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/states.py", line 91, in _get_state_result
    raise await get_state_exception(state)
  File "/usr/local/lib/python3.11/site-packages/prefect/engine.py", line 863, in orchestrate_flow_run
    result = await flow_call.aresult()
             ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/_internal/concurrency/calls.py", line 326, in aresult
    return await asyncio.wrap_future(self.future)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/_internal/concurrency/calls.py", line 388, in _run_async
    result = await coro
             ^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect_airbyte/flows.py", line 59, in run_connection_sync
    await task(airbyte_sync.wait_for_completion.aio)(airbyte_sync)
  File "/usr/local/lib/python3.11/site-packages/prefect/_internal/concurrency/api.py", line 182, in wait_for_call_in_loop_thread
    return call.result()
           ^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/_internal/concurrency/calls.py", line 317, in result
    return self.future.result(timeout=timeout)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/_internal/concurrency/calls.py", line 178, in result
    return self.__get_result()
           ^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/concurrent/futures/_base.py", line 401, in __get_result
    raise self._exception
  File "/usr/local/lib/python3.11/site-packages/prefect/_internal/concurrency/calls.py", line 388, in _run_async
    result = await coro
             ^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/engine.py", line 1568, in get_task_call_return_value
    return await future._result()
           ^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/futures.py", line 237, in _result
    return await final_state.result(raise_on_failure=raise_on_failure, fetch=True)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/states.py", line 91, in _get_state_result
    raise await get_state_exception(state)
  File "/usr/local/lib/python3.11/site-packages/prefect/engine.py", line 2107, in orchestrate_task_run
    result = await call.aresult()
             ^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/_internal/concurrency/calls.py", line 326, in aresult
    return await asyncio.wrap_future(self.future)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/_internal/concurrency/calls.py", line 388, in _run_async
    result = await coro
             ^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect_airbyte/connections.py", line 236, in wait_for_completion
    job_info = await airbyte_client.get_job_info(self.job_id)
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect_airbyte/client.py", line 227, in get_job_info
    raise err.AirbyteServerNotHealthyException() from e
prefect_airbyte.exceptions.AirbyteServerNotHealthyException
06:38:48 PM
prefect.flow_runs
Finished in state Failed('Flow run encountered an exception. AirbyteServerNotHealthyException: ')
06:38:48 PM
prefect.flow_runs
Process for flow run 'shrewd-hedgehog' exited cleanly.
06:38:49 PM
prefect.flow_runs.runner
m
Your message was too long by 2216 tokens - please shorten it and try again. For reference, here's your message at the allowed limit:
Hi
I’m using the 
prefect-airbyte
library to sync Airbyte Connections. Airbyte is running on Kubernetes (AKS). I’ve faced an error when Airbyte restarts it’s server or worker pods leading the Prefect flow failures. The airbyte sync continues to run. Below is the log. Can you advise how to debug further or fix it? ```Runner 'elt-deployment' submitting flow run '0177e1fb-a1a4-49e5-a2d7-62f5ef433ab7' 054110 PM prefect.flow_runs.runner Opening process... 054110 PM prefect.flow_runs.runner Completed submission of flow run '0177e1fb-a1a4-49e5-a2d7-62f5ef433ab7' 054110 PM prefect.flow_runs.runner Downloading flow code from storage at '.' 054112 PM prefect.flow_runs Created subflow run 'hospitable-goat' for flow 'airbyte_flow' 054112 PM prefect.flow_runs Encountered exception during execution: Traceback (most recent call last): File "/usr/local/lib/python3.11/site-packages/prefect_airbyte/client.py", line 220, in get_job_info response.raise_for_status() File "/usr/local/lib/python3.11/site-packages/http
e
Copy code
httpx.HTTPStatusError: Server error '500 Internal Server Error' for url '<http://airbyte-airbyte-server-svc.airbyte.svc.cluster.local:8001/api/v1/jobs/get/>'

File "/usr/local/lib/python3.11/site-packages/prefect_airbyte/client.py", line 227, in get_job_info
    raise err.AirbyteServerNotHealthyException() from e
prefect_airbyte.exceptions.AirbyteServerNotHealthyException
06:38:48 PM
prefect.flow_runs
Finished in state Failed('Flow run encountered an exception. AirbyteServerNotHealthyException: ')
@Marvin please see the logs
m
Ah, well, it seems my capabilities don't quite extend to viewing logs directly, especially without knowing what logs you're referring to. However, if you're looking for logs within a Prefect context or troubleshooting an issue related to Prefect, I can help you search the documentation or GitHub for relevant information or examples. Could you please provide me with more details or clarify what specific logs or issues you are interested in?
e
@Marvin these are my errors httpx.HTTPStatusError: Server error ‘500 Internal Server Error’ for url ‘http://airbyte-airbyte-server-svc.airbyte.svc.cluster.local:8001/api/v1/jobs/get/’ File “/usr/local/lib/python3.11/site-packages/prefect_airbyte/client.py”, line 227, in get_job_info raise err.AirbyteServerNotHealthyException() from e prefect_airbyte.exceptions.AirbyteServerNotHealthyException 063848 PM prefect.flow_runs Finished in state Failed(‘Flow run encountered an exception. AirbyteServerNotHealthyException: ’)
m
It seems you've encountered a couple of errors related to Airbyte and handling HTTP requests. 1. The 
httpx.HTTPStatusError: Server error ‘500 Internal Server Error’
implies that there was a server-side issue at the URL you were trying to reach. It indicates that Airbyte's server returned an HTTP 500 error, which is a generic error response for an unexpected condition encountered by the server. The Prefect documentation on handling HTTP requests with 
httpx
can be found here, though it does not directly address server errors. 2. The 
prefect_airbyte.exceptions.AirbyteServerNotHealthyException
is an exception indicating that the Airbyte server is not in a healthy state and therefore the job failed. It's raised directly by Prefect's Airbyte client when it's unable to retrieve job information due to the server's ill health. Unfortunately, I could not find direct documentation related to this specific Airbyte exception. It would be advisable to check the health of your Airbyte server and ensure it is operational. I also tried to search for any GitHub issues related to these errors, but there was no response that could be related to them directly at this time. To investigate further, consider: - Checking the Airbyte server's logs to diagnose the source of the 500 error. - Verifying that the Airbyte server is running and reachable from Prefect's environment. - Ensuring that network configurations, such as firewall rules and DNS settings, allow for communication between Prefect and Airbyte services. For more detailed advice and specific troubleshooting, it may be beneficial to reach out to the Prefect Community Slack or Airbyte's community for support.
e
@Marvin is there a way to retry the airbyte connection if there is an unhealthy exception? using the 
prefect-airbyte
library
m
To retry the Airbyte connection in case of an 
AirbyteServerNotHealthyException
, you can implement a retry mechanism within your Prefect flow using tasks. Prefect provides utility functions for implementing retry strategies, like an exponential backoff, which can be useful in this case. You can define a custom Prefect task that encapsulates the Airbyte connection logic and makes use of the built-in support for retries. Here is an example pattern you might use (adapt it to your specific Airbyte task):
Copy code
python
from prefect import flow, task, get_retry_delay
from prefect.tasks.prefect_airbyte import AirbyteConnectionTask
from prefect_airbyte.exceptions import AirbyteServerNotHealthyException
from datetime import timedelta

@task(retries=3, retry_delay=get_retry_delay.exponential_backoff)
def safe_airbyte_sync(connection_id: str):
    try:
        airbyte_task = AirbyteConnectionTask()
        airbyte_task.run(connection_id=connection_id)
    except AirbyteServerNotHealthyException as e:
        # The task will be retried in case of AirbyteServerNotHealthyException
        raise e

@flow
def my_flow():
    # Replace 'your_connection_id' with the actual Airbyte connection ID
    safe_airbyte_sync(connection_id='your_connection_id')

# To run the flow
my_flow()
In this example: - 
retries
is set to 3, which means Prefect will attempt to run the task up to 4 times in total (1 initial run + 3 retries). - 
retry_delay
uses the 
get_retry_delay.exponential_backoff
function to define the delay between retries. - 
AirbyteConnectionTask()
is the task from the 
prefect-airbyte
library that triggers an Airbyte connection. - 
safe_airbyte_sync
catches the 
AirbyteServerNotHealthyException
and raises it, allowing Prefect to handle the retry logic. For more detailed documentation on tasks with Prefect, you can visit the Prefect Tasks documentation. Moreover, to see more examples of Prefect code including how to define flows and tasks, you can access a basic Prefect flow example here. Please adjust the code samples provided to fit the specific needs of your application and integration with Airbyte.
e
@Eric Sales De Andrade did you figure this out? Hitting the same exact error myself. I think the health check in the sync is using an outdated API endpoint address
e
I implemented my own health check mechanism and switched to using tasks with retries instead of subflows
22 Views
Open in Slack
PreviousNext
Powered by