Airflow Dynamic Task Fails with httpx time out on an Instance with high load and around 300 active DAGs.

[saaifali]

Airflow version 3.1.1
Running on Kubernetes in a Corporate Network
Active DAGs : Around 300
Deployment:
2 Api servers with one worker each
1 DAG Processor
8 Schedulers
1 Triggerer
Each has around 4 CPU and 20 GB memory limits.

I have a particular DAG set up with Dynamic Task Mapping which also uses the .map function in between to map the output from my first task to have the proper format for the dynamic mapped tasks.
Now this DAG ran perfectly fine on a dev server with less number of DAGs and low load.

When I moved it to test in Production, My setup task which is responsible for deciding which dynamic mapped task to run completes successfully, but the two following jobs (one running on KubernetesExecutor and the other on LocalExecutor) keeps on failing at the same step.
This is the Part where it tries to fetch the Xcoms from the API server and times out with an Httpx Error : timeout.

The relevant DAG code:

# Step 1 — determine which job (if any) should run this DAG run.
    #           Returns a list of 0 or 1 job-descriptor dicts.
    job_list = cgs_jobs_scheduler()

    k8s_cmd = job_list.map(_k8s_cmd_filter)
    local_cmd = job_list.map(_local_cmd_filter)
    # Step 3a — .map() extracts the bash_command for k8s jobs only.
    #            If the job is local, _k8s_cmd_filter raises AirflowSkipException
    #            → that mapped instance is skipped.
    #            0 items in job_list → 0 mapped instances (task skipped entirely).
    k8s_job = BashOperator.partial(
        task_id='k8s_job',
        #map_index_template="""{{ task.bash_command.split('/')[-1].replace('.sh', '') }}""",
        priority_weight=1,
        env={
            "AIRFLOW_TASK_URL": "{{ task_instance.log_url }}",
            "XDV_USER": "{{ params.jobs_to_trigger[2] if params.jobs_to_trigger|length > 2 else '' }}",
            "WATERFALL_TYPE": "{{ params.WATERFALL_TYPE if 'WATERFALL_TYPE' in params else '' }}",
            "DV_RELEASE_KEY": "{{ params.DV_RELEASE_KEY if 'DV_RELEASE_KEY' in params else '' }}",
        },
        append_env=True,
        executor='KubernetesExecutor',
    ).expand(bash_command=k8s_cmd)

    # Step 3b — .map() extracts the bash_command for local jobs only.
    #            If the job is k8s, _local_cmd_filter raises AirflowSkipException
    #            → that mapped instance is skipped.
    local_job = BashOperator.partial(
        task_id='local_job',
        #map_index_template="""{{ task.bash_command.split('/')[-1].replace('.sh', '') }}""",
        priority_weight=1,
        env={
            "AIRFLOW_TASK_URL": "{{ task_instance.log_url }}",
            "XDV_USER": "{{ params.jobs_to_trigger[2] if params.jobs_to_trigger|length > 2 else '' }}",
            "WATERFALL_TYPE": "{{ params.WATERFALL_TYPE if 'WATERFALL_TYPE' in params else '' }}",
            "DV_RELEASE_KEY": "{{ params.DV_RELEASE_KEY if 'DV_RELEASE_KEY' in params else '' }}",
        },
        append_env=True,
        executor='LocalExecutor',
    ).expand(bash_command=local_cmd)

The error in question:

Starting call to 'airflow.sdk.api.client.Client.request', this is the 4th time calling it.
[2026-04-02T17:38:35.849180Z] {retries.py:99} DEBUG - Running DagRun.get_running_dag_runs_to_examine with retries. Try 1 of 5
[2026-04-02T17:38:37.867007Z] {local_executor.py:100} ERROR - uhoh
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.11/site-packages/httpx/_transports/default.py", line 101, in map_httpcore_exceptions
    yield
  File "/home/airflow/.local/lib/python3.11/site-packages/httpx/_transports/default.py", line 250, in handle_request
    resp = self._pool.handle_request(req)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.11/site-packages/httpcore/_sync/connection_pool.py", line 256, in handle_request
    raise exc from None
  File "/home/airflow/.local/lib/python3.11/site-packages/httpcore/_sync/connection_pool.py", line 236, in handle_request
    response = connection.handle_request(
               ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.11/site-packages/httpcore/_sync/connection.py", line 103, in handle_request
    return self._connection.handle_request(request)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.11/site-packages/httpcore/_sync/http11.py", line 136, in handle_request
    raise exc
  File "/home/airflow/.local/lib/python3.11/site-packages/httpcore/_sync/http11.py", line 106, in handle_request
    ) = self._receive_response_headers(**kwargs)
        ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.11/site-packages/httpcore/_sync/http11.py", line 177, in _receive_response_headers
    event = self._receive_event(timeout=timeout)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.11/site-packages/httpcore/_sync/http11.py", line 217, in _receive_event
    data = self._network_stream.read(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.11/site-packages/httpcore/_backends/sync.py", line 126, in read
    with map_exceptions(exc_map):
  File "/usr/python/lib/python3.11/contextlib.py", line 158, in __exit__
    self.gen.throw(typ, value, traceback)
  File "/home/airflow/.local/lib/python3.11/site-packages/httpcore/_exceptions.py", line 14, in map_exceptions
    raise to_exc(exc) from exc
httpcore.ReadTimeout: timed out
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.11/site-packages/airflow/executors/local_executor.py", line 96, in _run_worker
    _execute_work(log, workload)
  File "/home/airflow/.local/lib/python3.11/site-packages/airflow/executors/local_executor.py", line 124, in _execute_work
    supervise(
  File "/home/airflow/.local/lib/python3.11/site-packages/airflow/sdk/execution_time/supervisor.py", line 1935, in supervise
    exit_code = process.wait()
                ^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.11/site-packages/airflow/sdk/execution_time/supervisor.py", line 990, in wait
    self._monitor_subprocess()
  File "/home/airflow/.local/lib/python3.11/site-packages/airflow/sdk/execution_time/supervisor.py", line 1055, in _monitor_subprocess
    alive = self._service_subprocess(max_wait_time=max_wait_time) is None
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.11/site-packages/airflow/sdk/execution_time/supervisor.py", line 771, in _service_subprocess
    need_more = socket_handler(key.fileobj)
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.11/site-packages/airflow/sdk/execution_time/supervisor.py", line 1713, in cb
    gen.send(request)
  File "/home/airflow/.local/lib/python3.11/site-packages/airflow/sdk/execution_time/supervisor.py", line 612, in handle_requests
    self._handle_request(msg, log, request.id)
  File "/home/airflow/.local/lib/python3.11/site-packages/airflow/sdk/execution_time/supervisor.py", line 1216, in _handle_request
    xcom = self.client.xcoms.get(
           ^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.11/site-packages/airflow/sdk/api/client.py", line 448, in get
    resp = self.client.get(f"xcoms/{dag_id}/{run_id}/{task_id}/{key}", params=params)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.11/site-packages/httpx/_client.py", line 1053, in get
    return self.request(
           ^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.11/site-packages/tenacity/__init__.py", line 338, in wrapped_f
    return copy(f, *args, **kw)
           ^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.11/site-packages/tenacity/__init__.py", line 477, in __call__
    do = self.iter(retry_state=retry_state)
         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.11/site-packages/tenacity/__init__.py", line 378, in iter
    result = action(retry_state)
             ^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.11/site-packages/tenacity/__init__.py", line 420, in exc_check
    raise retry_exc.reraise()
          ^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.11/site-packages/tenacity/__init__.py", line 187, in reraise
    raise self.last_attempt.result()
          ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/python/lib/python3.11/concurrent/futures/_base.py", line 449, in result
    return self.__get_result()
           ^^^^^^^^^^^^^^^^^^^
  File "/usr/python/lib/python3.11/concurrent/futures/_base.py", line 401, in __get_result
    raise self._exception
  File "/home/airflow/.local/lib/python3.11/site-packages/tenacity/__init__.py", line 480, in __call__
    result = fn(*args, **kwargs)
             ^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.11/site-packages/airflow/sdk/api/client.py", line 866, in request
    return super().request(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.11/site-packages/httpx/_client.py", line 825, in request
    return self.send(request, auth=auth, follow_redirects=follow_redirects)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.11/site-packages/httpx/_client.py", line 914, in send
    response = self._send_handling_auth(
               ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.11/site-packages/httpx/_client.py", line 942, in _send_handling_auth
    response = self._send_handling_redirects(
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.11/site-packages/httpx/_client.py", line 979, in _send_handling_redirects
    response = self._send_single_request(request)
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.11/site-packages/httpx/_client.py", line 1014, in _send_single_request
    response = transport.handle_request(request)
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.11/site-packages/httpx/_transports/default.py", line 249, in handle_request
    with map_httpcore_exceptions():
  File "/usr/python/lib/python3.11/contextlib.py", line 158, in __exit__
    self.gen.throw(typ, value, traceback)
  File "/home/airflow/.local/lib/python3.11/site-packages/httpx/_transports/default.py", line 118, in map_httpcore_exceptions
    raise mapped_exc(message) from exc
httpx.ReadTimeout: timed out

I am at my wits end on how to resolve this issue.
Any ideas, suggestions, prayers, spells, etc will be appreciated. 💯

[Urus1201]

The root cause is that your API servers are heavily under-provisioned for a 300-DAG, 8-scheduler setup. The Airflow 3 task SDK communicates with the API server for XCom reads (and other task operations) — when the API server is saturated, those calls time out.

Why this works on dev but not prod

In Airflow 3, tasks fetch XComs by calling the REST API (/xcoms/{dag_id}/{run_id}/{task_id}/{key}) rather than reading from the metastore directly. With 8 schedulers and 300 active DAGs all running tasks simultaneously, your 2 API server pods with 1 worker each = 2 total worker processes handling all API traffic. That is far too few.

Fix 1: Increase API server workers (highest impact)

In your Helm chart values:

apiServer:
  replicas: 2
  workers: 4   # Increase from 1 to 4 (or more)

Or if using gunicorn-based config:

[api_server]
workers = 4
worker_timeout = 120

This alone will likely resolve the timeout.

Fix 2: Increase the SDK client timeout

Set a higher timeout for the task SDK's HTTP client:

[api]
# Timeout in seconds for task SDK → API server calls
client_connect_timeout = 10
client_read_timeout = 60   # default is often too low under load

Fix 3: Check your corporate network proxy

The stack trace shows the timeout happening at the TCP read level (httpcore._sync.http11), which can be caused by a proxy or firewall inserting latency on pod-to-pod HTTP calls. If your workers and API servers communicate through a corporate proxy, bypass it for internal traffic:

env:
  - name: NO_PROXY
    value: "localhost,127.0.0.1,.cluster.local,<api-server-service-name>"

Fix 4: Reduce XCom payload size

Your dynamic task uses .map() which generates XCom entries per mapped instance. Large XCom payloads (lists of job descriptors) passed between tasks increase API server response time. Consider:

• Using an XCom backend (S3/GCS) for large payloads
• Passing only IDs/keys between tasks, fetching the full data inside each task

TL;DR

Change	Expected impact
Increase API workers from 1 to 4+	Eliminates timeout under load
Raise client_read_timeout	Prevents timeout for slow-but-valid responses
Add NO_PROXY for internal traffic	Fixes proxy-induced latency
Reduce XCom payload size	Reduces API server load long-term