(airflow) branch v3-4-test updated: [v3-4-test] Fix XComs of task-end trigger events being JSON-encoded twice (#76596) (#76989)

[shahar.apache.org via commits]

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

shahar1 pushed a commit to branch v3-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git


The following commit(s) were added to refs/heads/v3-4-test by this push:
     new ad9973e7172 [v3-4-test] Fix XComs of task-end trigger events being 
JSON-encoded twice (#76596) (#76989)
ad9973e7172 is described below

commit ad9973e7172a943b6606adebdf706769aab66a8c
Author: github-actions[bot] 
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Thu Nov 12 21:36:05 2026 +0000

    [v3-4-test] Fix XComs of task-end trigger events being JSON-encoded twice 
(#76596) (#76989)
    
    (cherry picked from commit 1c870b3478d63680151b0170f6427be8756a11ce)
    
    Co-authored-by: raphaelauv <raphael...@gmail.com>
---
 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 656a2a474a1..7a504ff7f14 100644
--- a/airflow-core/src/airflow/models/trigger.py
+++ b/airflow-core/src/airflow/models/trigger.py
@@ -745,9 +745,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 ff12adf0f37..43923fa21ab 100644
--- a/airflow-core/tests/unit/models/test_trigger.py
+++ b/airflow-core/tests/unit/models/test_trigger.py
@@ -672,7 +672,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()
@@ -683,10 +685,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")