(airflow) branch main updated: Fix XComs of task-end trigger events being JSON-encoded twice (#76596)

[shahar.apache.org via commits]

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

shahar1 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 1c870b3478d Fix XComs of task-end trigger events being JSON-encoded 
twice (#76596)
1c870b3478d is described below

commit 1c870b3478d63680151b0170f6427be8756a11ce
Author: raphaelauv <raphael...@gmail.com>
AuthorDate: Thu Nov 12 20:01:27 2026 +0100

    Fix XComs of task-end trigger events being JSON-encoded twice (#76596)
---
 airflow-core/src/airflow/models/trigger.py     | 10 +++++++++-
 airflow-core/tests/unit/models/test_trigger.py |  9 ++++-----
 2 files changed, 13 insertions(+), 6 deletions(-)

diff --git a/airflow-core/src/airflow/models/trigger.py 
b/airflow-core/src/airflow/models/trigger.py
index de6467a5851..a095e4ef9fa 100644
--- a/airflow-core/src/airflow/models/trigger.py
+++ b/airflow-core/src/airflow/models/trigger.py
@@ -953,9 +953,17 @@ def _(event: BaseTaskEndEvent, *, task_instance: 
TaskInstance, session: Session)
 
     def _push_xcoms_if_necessary() -> None:
         """Pushes XComs to the database if they are provided."""
+        from airflow.models.xcom import XComModel
+
         if event.xcoms and callback_type != TaskInstanceState.UP_FOR_RETRY:
             for key, value in event.xcoms.items():
-                task_instance.xcom_push(key=key, value=value)
+                XComModel.set_for_attempt(
+                    task_instance_id=task_instance.id,
+                    key=key,
+                    value=value,
+                    serialize=False,
+                    session=session,
+                )
 
     # Send the callback before mutating task state so it reflects the 
retry-vs-terminal
     # decision derived above.
diff --git a/airflow-core/tests/unit/models/test_trigger.py 
b/airflow-core/tests/unit/models/test_trigger.py
index 5963f18fa62..7908cc268c4 100644
--- a/airflow-core/tests/unit/models/test_trigger.py
+++ b/airflow-core/tests/unit/models/test_trigger.py
@@ -871,7 +871,9 @@ def test_submit_event_task_end(mock_utcnow, session, 
create_task_instance, event
     # now, for each type, submit event
     # verify that (1) task ends in right state and (2) xcom is pushed
     Trigger.submit_event(
-        trigger.id, event_cls(xcoms={"return_value": "xcomret", "a": "b", "c": 
"d"}), session=session
+        trigger.id,
+        event_cls(xcoms={"return_value": "xcomret", "a": "b", "c": {"d": [1, 
2]}}),
+        session=session,
     )
     # commit changes made by submit event and expire all cache to read from db.
     session.flush()
@@ -882,10 +884,7 @@ def test_submit_event_task_end(mock_utcnow, session, 
create_task_instance, event
     assert ti.end_date == now
     assert ti.duration is not None
     actual_xcoms = {x.key: x.value for x in get_xcoms(ti)}
-    expected_xcoms = {}
-    for k, v in {"return_value": "xcomret", "a": "b", "c": "d"}.items():
-        expected_xcoms[k] = json.dumps(v)
-    assert actual_xcoms == expected_xcoms
+    assert actual_xcoms == {"return_value": "xcomret", "a": "b", "c": {"d": 
[1, 2]}}
 
 
 @patch("airflow.callbacks.database_callback_sink.DatabaseCallbackSink.send")