This is an automated email from the ASF dual-hosted git repository.
kaxil pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new f05c32e10e1 Lower severity of handled XCom misses (#73198)
f05c32e10e1 is described below
commit f05c32e10e1a37289175ebd4c61b647252737960
Author: AxelRay <[email protected]>
AuthorDate: Thu Nov 5 05:02:15 2026 +0700
Lower severity of handled XCom misses (#73198)
* Lower log level for handled XCom misses
* Log handled XCom misses at debug level
LazyXComIterator reads offsets until the server returns 404, so every
iteration over a mapped upstream ends in a handled miss. Log both XCom
not-found branches at debug, like the Connection and Variable lookups.
Move the new tests to httpx2, use the server's real 404 body, and name
them after the fallback they cover.
---------
Co-authored-by: Kaxil Naik <[email protected]>
---
task-sdk/src/airflow/sdk/api/client.py | 4 +--
task-sdk/tests/task_sdk/api/test_client.py | 50 ++++++++++++++++++++++++++++++
2 files changed, 52 insertions(+), 2 deletions(-)
diff --git a/task-sdk/src/airflow/sdk/api/client.py
b/task-sdk/src/airflow/sdk/api/client.py
index bdf5ba34bff..2638a2a697c 100644
--- a/task-sdk/src/airflow/sdk/api/client.py
+++ b/task-sdk/src/airflow/sdk/api/client.py
@@ -659,7 +659,7 @@ class XComOperations:
resp = self.client.get(f"xcoms/{dag_id}/{run_id}/{task_id}/{key}",
params=params)
except ServerResponseError as e:
if e.response.status_code == HTTPStatus.NOT_FOUND:
- log.error(
+ log.debug(
"XCom not found",
dag_id=dag_id,
run_id=run_id,
@@ -740,7 +740,7 @@ class XComOperations:
)
except ServerResponseError as e:
if e.response.status_code == HTTPStatus.NOT_FOUND:
- log.error(
+ log.debug(
"XCom not found",
dag_id=dag_id,
run_id=run_id,
diff --git a/task-sdk/tests/task_sdk/api/test_client.py
b/task-sdk/tests/task_sdk/api/test_client.py
index 9b23b65c153..a0bd43dffbb 100644
--- a/task-sdk/tests/task_sdk/api/test_client.py
+++ b/task-sdk/tests/task_sdk/api/test_client.py
@@ -1163,6 +1163,56 @@ class TestXCOMOperations:
key="key",
)
+ def test_xcom_get_404_returns_none_value(self):
+ def handle_request(request: httpx2.Request) -> httpx2.Response:
+ if request.url.path == "/xcoms/dag_id/run_id/task_id/key":
+ return httpx2.Response(
+ status_code=404,
+ json={"detail": {"reason": "not_found", "message": "XCom
not found"}},
+ )
+ return httpx2.Response(status_code=400, json={"detail": "Bad
Request"})
+
+ client = make_client(transport=httpx2.MockTransport(handle_request))
+
+ result = client.xcoms.get(
+ dag_id="dag_id",
+ run_id="run_id",
+ task_id="task_id",
+ key="key",
+ )
+
+ assert result == XComResponse(key="key", value=None)
+
+ def test_xcom_get_sequence_item_404_returns_not_found_error(self):
+ def handle_request(request: httpx2.Request) -> httpx2.Response:
+ if request.url.path == "/xcoms/dag_id/run_id/task_id/key/item/2":
+ return httpx2.Response(
+ status_code=404,
+ json={"detail": {"reason": "not_found", "message": "XCom
not found"}},
+ )
+ return httpx2.Response(status_code=400, json={"detail": "Bad
Request"})
+
+ client = make_client(transport=httpx2.MockTransport(handle_request))
+
+ result = client.xcoms.get_sequence_item(
+ dag_id="dag_id",
+ run_id="run_id",
+ task_id="task_id",
+ key="key",
+ offset=2,
+ )
+
+ assert result == ErrorResponse(
+ error=ErrorType.XCOM_NOT_FOUND,
+ detail={
+ "dag_id": "dag_id",
+ "run_id": "run_id",
+ "task_id": "task_id",
+ "key": "key",
+ "offset": 2,
+ },
+ )
+
@pytest.mark.parametrize(
"values",
[
