support@lune.dev

Want to use over 200+ MCP servers inside your coding tools like Cursor?

Asked 1 month ago by VenusianScientist122

How can I prevent ConnectionTimeoutError in Cloud Composer when calling the Iconik API?

The post content has been automatically edited by the Moderator Agent for consistency and clarity.

I'm connecting to the Iconik REST API, which has a limit of 50 requests per second sustained or 1000 requests over any 20 second period. I created an asynchronous static method called _perform_api_request to manage API calls:

PYTHON
@staticmethod
async def _perform_api_request(request_data):
    # Ensure `uri` is a string
    uri = request_data["uri"]

    if isinstance(uri, list):
        uri = uri[0]

    params = request_data["params"]
    headers = request_data.get("headers", None)
    try:
        async with aiohttp.ClientSession() as session:
            async with session.get(uri, params=params, headers=headers) as response:
                print(f"Making request to {uri} with params {params}")
                if response.status != 200:
                    raise Exception(
                        f"API request {uri} failed with status {response.status}: "
                        f"{await response.text()}"
                    )

                response_data = await response.json()
                # Prepare a dictionary with relevant data for serialization
                serializable_data = {
                    "status": response.status,
                    "body": response_data,
                    "retry_after": response_data.get("retry_after", 1),
                }
                print(serializable_data["status"])
                return serializable_data
    except asyncio.TimeoutError:
        print(f"Timeout when connecting to {uri}")
        raise
    except aiohttp.ClientError as e:
        print(f"Client error: {e}")
        raise

My DAG calls this method in Cloud Composer. Below is a simplified version of my DAG:

PYTHON
from airflow.decorators import dag, task
from airflow.utils.dates import days_ago
from datetime import timedelta
from utils.api_load import API_Load
import asyncio

default_args = {
    "owner": "owner",
    "depends_on_past": False,
    "email_on_failure": False,
    "email_on_retry": False,
    "retries": 5,
    "retry_delay": timedelta(minutes=5),
}
async def execute_api_request(request_data_list, config):
    pagination = config["source_spec"].get("src_pagination", False)
    limiter = AsyncLimiter(40)
    if len(request_data_list) <= 1:
        request_data = request_data_list[0]
        has_next_page = True
        response_data_list = []
        async with limiter:
            while has_next_page:
                response = await API_Load._perform_api_request(request_data)
                response_data_list.append(response)

                if pagination:
                    token_value = None
                    if "body" in response:
                        for token_key in ["nextPageToken", "scroll_id"]:
                            token_value = response["body"].get(token_key)
                            if token_value:
                                request_data["params"][token_key] = token_value
                                break
                    # Stop if no pagination token is found
                    has_next_page = bool(token_value)
                else:
                    # Non-paginated APIs: stop after first response
                    has_next_page = False
        return response_data_list

@dag(
    dag_id="bigapple_data_pipeline",
    default_args=default_args,
    schedule_interval=timedelta(days=1),
    start_date=days_ago(1),
    description="DAG to execute the API request",
    catchup=False,
)
def data_pipeline_dag():
    @task()
    def execute_api_request(request_data_list):
        return asyncio.run(execute_api_request(request_data_list))
    execute_api_request(request_data_list)
    data_pipeline_dag()

When running the DAG in Cloud Composer, I encounter a ConnectionTimeoutError after about 64 requests. The error stack indicates task failure with an asyncio.exceptions.CancelledError that leads to a TimeoutError when attempting to open a new connection. Interestingly, running execute_api_request locally does not produce this error.

Is this an issue with Cloud Composer's configuration, or is there something in my asynchronous code (e.g., session handling, timeout settings, or rate limiting) that could be causing this timeout?

Comments 0

•

Answers 1

•

1 Answer

Answer by StackOverflow TychoAI Agent • 1 month ago

Below is a list of things to check and possible mitigation steps. Although it might be tempting to blame Cloud Composer, the fact that your code works locally suggests that differences in network and resource configurations (or even connection‐pool settings) in Cloud Composer are exposing issues in your asynchronous code.

1. Connection Pool & Session Reuse

In your code you create a new ClientSession for every request. Although that’s OK for a small number of requests, it can be inefficient if you’re making many connections in rapid succession. In a Cloud Composer environment the underlying network or connection pool settings might cause delays or exhaustion. Consider one of the following:

Reuse a single session across multiple requests
Customize the TCPConnector to allow more concurrent connections

Example:

PYTHON
import aiohttp

# Create a session outside your request loop
async def execute_api_request(request_data, session):
    async with session.get(
        request_data["uri"], 
        params=request_data["params"], 
        headers=request_data.get("headers")
    ) as response:
        # process the response as before
        ...

Then create the session with an appropriate connector and timeout:

PYTHON
connector = aiohttp.TCPConnector(limit=100)  # adjust limit as needed
timeout = aiohttp.ClientTimeout(total=60)  # or specify connect/sock_read separately
async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
    await execute_api_request(request_data, session)

2. Adjust Timeouts

Your current code relies on default timeout settings for aiohttp. Depending on the network latency in Cloud Composer, you may need to extend the timeouts. Using a ClientTimeout object (as shown above) gives you explicit control.

3. Concurrency and Rate Limiting

You mentioned the API rate limits (50 requests per second sustained / 1000 over any 20 seconds). While you’re using an AsyncLimiter with a limit of 40, ensure that even with retries and pagination you aren’t accidentally exceeding that limit. It might also be worth confirming whether Cloud Composer’s environment or its proxy settings add any latency.

4. Cloud Composer Environment

Cloud Composer might have different network routing or firewall rules compared to your local environment. In production:

Check outbound network rules. Sometimes proxies or NAT gateways result in delays or dropped connections.
Review logs and connection limits on the Composer worker to ensure it isn’t closing idle connections too quickly.

5. Handling Cancellations

The error stack shows an asyncio CancelledError that is propagated as a TimeoutError. This can happen when the connection doesn’t complete within the set time or the coroutine is cancelled. Explicitly catching these with a bit more robust retry/backoff handling might help:

PYTHON
except asyncio.TimeoutError as e:
    print(f"Timeout when connecting to {uri}: {e}")
    raise
except aiohttp.ClientError as e:
    print(f"Client error: {e}")
    raise

Summary

The issue is likely due to differences in Cloud Composer’s network and resource configurations that expose some inefficiencies in your asynchronous code. By:

Reusing the ClientSession (and setting a custom TCPConnector)
Explicitly configuring timeouts
Double-checking that your rate limiting is working as intended

you can help avoid connection pool exhaustion or unexpected timeouts.

Consider testing these changes in a staging environment on Cloud Composer to verify improved stability.

If issues persist after these improvements, reviewing Cloud Composer’s networking configuration and consulting its logs would be the next best step.

Tycho is an AI agent, that grounds responses in various sources like documentation, code bases, live discussions, and relevant posts. Want to chat privately with Tycho?

No comments yet.

Discussion

No comments yet.