(airflow) branch main updated: Prevent FileNotFoundError in deserialize_value of XCom io provider to crash the scheduler (#52702)

[po...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

potiuk 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 ad10a103f76 Prevent FileNotFoundError in deserialize_value of XCom io provider to crash the scheduler (#52702)
ad10a103f76 is described below

commit ad10a103f760d12993f5c1112f8960e9d3cb99b9
Author: David Blain <in...@dabla.be>
AuthorDate: Fri Jul 4 15:31:18 2025 +0200

    Prevent FileNotFoundError in deserialize_value of XCom io provider to crash the scheduler (#52702)
    
    * refactor: Prevent FileNotFoundError in deserialize_value of XCom backend provider to crash the scheduler
    
    * refactor: Reformatted test_serialization_deserialization_basic
    
    * refactor: Made import of ObjectStoragePath cross airflow version compatible
    
    ---------
    
    Co-authored-by: David Blain <da...@infrabel.be>
---
 .../airflow/providers/common/io/xcom/backend.py    |  2 +-
 .../io/tests/unit/common/io/xcom/test_backend.py   | 37 ++++++++++++++++------
 2 files changed, 29 insertions(+), 10 deletions(-)

diff --git a/providers/common/io/src/airflow/providers/common/io/xcom/backend.py b/providers/common/io/src/airflow/providers/common/io/xcom/backend.py
index 4dd868067b8..01211b706ad 100644
--- a/providers/common/io/src/airflow/providers/common/io/xcom/backend.py
+++ b/providers/common/io/src/airflow/providers/common/io/xcom/backend.py
@@ -173,7 +173,7 @@ class XComObjectStorageBackend(BaseXCom):
         try:
             with path.open(mode="rb", compression="infer") as f:
                 return json.load(f, cls=XComDecoder)
-        except (TypeError, ValueError):
+        except (FileNotFoundError, TypeError, ValueError):
             return data
 
     @staticmethod
diff --git a/providers/common/io/tests/unit/common/io/xcom/test_backend.py b/providers/common/io/tests/unit/common/io/xcom/test_backend.py
index 08967ca32d2..f85a4cfa0c7 100644
--- a/providers/common/io/tests/unit/common/io/xcom/test_backend.py
+++ b/providers/common/io/tests/unit/common/io/xcom/test_backend.py
@@ -17,7 +17,7 @@
 # under the License.
 from __future__ import annotations
 
-from unittest.mock import MagicMock
+from unittest.mock import MagicMock, patch
 
 import pytest
 
@@ -35,9 +35,11 @@ pytestmark = [pytest.mark.db_test]
 
 if AIRFLOW_V_3_0_PLUS:
     from airflow.models.xcom import XComModel
+    from airflow.sdk import ObjectStoragePath
     from airflow.sdk.execution_time.comms import XComResult
     from airflow.sdk.execution_time.xcom import resolve_xcom_backend
 else:
+    from airflow.io.path import ObjectStoragePath
     from airflow.models.xcom import BaseXCom, resolve_xcom_backend  # type: ignore[no-redef]
 
 
@@ -373,6 +375,11 @@ class TestXComObjectStorageBackend:
     @pytest.mark.parametrize(
         "value, expected_value",
         [
+            pytest.param(
+                "file://airflow/xcoms/non_existing_file.json",
+                "file://airflow/xcoms/non_existing_file.json",
+                id="str",
+            ),
             pytest.param(1, 1, id="int"),
             pytest.param(1.0, 1.0, id="float"),
             pytest.param("string", "string", id="str"),
@@ -385,11 +392,23 @@ class TestXComObjectStorageBackend:
         ],
     )
     def test_serialization_deserialization_basic(self, value, expected_value):
-        XCom = resolve_xcom_backend()
-        airflow.models.xcom.XCom = XCom
-
-        serialized_data = XCom.serialize_value(value)
-        mock_xcom_ser = MagicMock(value=serialized_data)
-        deserialized_data = XCom.deserialize_value(mock_xcom_ser)
-
-        assert deserialized_data == expected_value
+        def conditional_side_effect(data) -> ObjectStoragePath:
+            if isinstance(data, str) and data.startswith("file://"):
+                return ObjectStoragePath(data)
+            return original_get_full_path(data)
+
+        original_get_full_path = XComObjectStorageBackend._get_full_path
+
+        with patch.object(
+            XComObjectStorageBackend,
+            "_get_full_path",
+            side_effect=conditional_side_effect,
+        ):
+            XCom = resolve_xcom_backend()
+            airflow.models.xcom.XCom = XCom
+
+            serialized_data = XCom.serialize_value(value)
+            mock_xcom_ser = MagicMock(value=serialized_data)
+            deserialized_data = XCom.deserialize_value(mock_xcom_ser)
+
+            assert deserialized_data == expected_value