xcom_pull logs at ERROR level for an expected "XCom not found" case in Airflow 3.x Task SDK

Start in airflow/sdk/api/client.py at XComOperations.get() and get_sequence_item(), focusing on the handled 404 paths that log “XCom not found.” Adjust the logging level for those expected cases while preserving the existing None-return behavior, then verify the relevant client tests and the optional-XCom workflow behavior.

Description

In Airflow 3.x, TaskInstance.xcom_pull() executed from inside a running task goes through airflow.sdk.api.client.py::XComOperations.get(). When the requested key/task_id doesn't have a matching XCom, the API server returns a 404, which is caught and handled gracefully — the call still returns None, matching Airflow 2.x behavior — but before returning, it logs:

log.error(
    "XCom not found",
    dag_id=dag_id, run_id=run_id, task_id=task_id, key=key,
    map_index=map_index, detail=e.detail, status_code=e.response.status_code,
)

This means every task that does a "check if this XCom exists" style xcom_pull (a very common and valid pattern — pulling with a default, or checking optional upstream output) emits a task-log ERROR line even though nothing failed and no exception propagates. This is confusing for on-call/monitoring setups that alert on ERROR-level task logs, and misrepresents an expected, handled condition as an error.

This behavior was introduced by #45344 (closing #45341), which fixed a real bug — task crashes on missing XCom — by catching the 404 and returning None. That PR's discussion focused entirely on not crashing; the log level wasn't discussed.

Suggested fix

Downgrade these to log.warning (or lower) in airflow/sdk/api/client.py, since the surrounding code explicitly treats the condition as expected/handled, not an error:

• XComOperations.get() (~line 535, "XCom not found")
• XComOperations.get_sequence_item() (~line 606, "XCom not found")

(Possibly also "Asset not found" at ~line 674 for the same reason, though that's a separate call site — happy to split if preferred.)

Workaround

[logging] namespace_levels = airflow.sdk.api.client=CRITICAL in airflow.cfg silences it, but that's an all-or-nothing hammer on the whole client logger (also silences genuine connectivity/auth errors from the same module).

Environment

Observed on Airflow 3.2.2, but the code path is unchanged as of main as far as I can tell.

Are you willing to submit a PR?

• Yes I am willing to submit a PR!