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 11174724838 Add the resume event to stored next_kwargs without
decoding them (#73576)
11174724838 is described below
commit 11174724838fd21cfe6d630d4fae9dd88c186992
Author: Jarek Potiuk <[email protected]>
AuthorDate: Mon Oct 19 16:56:42 2026 +0200
Add the resume event to stored next_kwargs without decoding them (#73576)
* Add the resume event to stored next_kwargs without decoding them
handle_event_submit only needs to add the trigger event to a task
instance's stored next_kwargs. It previously decoded the whole stored
payload, added the event and re-encoded everything, reconstructing every
stored value in the scheduler, triggerer and API server although only the
worker ever needs those objects.
Merge the serialized event into the stored serialized dict instead. Stored
values pass through untouched; payloads that are not a dict at the top
level still fail the task instance rather than resuming it. Rows still in
the legacy {"__type": "dict", "__var": ...} shape keep that shape.
Generated-by: Claude Opus 5
* Fail the task when a legacy resume event cannot be serialized
Generated-by: Claude Opus 5
---
airflow-core/src/airflow/models/trigger.py | 68 ++++++++++++++-----------
airflow-core/tests/unit/models/test_trigger.py | 70 ++++++++++++++++++++++++++
generated/known_sdk_imports_in_core.txt | 2 +-
3 files changed, 110 insertions(+), 30 deletions(-)
diff --git a/airflow-core/src/airflow/models/trigger.py
b/airflow-core/src/airflow/models/trigger.py
index 5431516d048..1713f9aebb7 100644
--- a/airflow-core/src/airflow/models/trigger.py
+++ b/airflow-core/src/airflow/models/trigger.py
@@ -538,32 +538,36 @@ class Trigger(Base):
return result
-def _decode_next_kwargs(next_kwargs_raw: Any) -> dict[str, Any]:
+def _split_stored_next_kwargs(next_kwargs_raw: Any) -> tuple[dict[str, Any],
bool]:
"""
- Decode the stored ``next_kwargs`` of a task instance into a plain dict.
+ Return the top-level mapping of the stored ``next_kwargs``, with its
values still encoded.
- Deserialize with serde first to provide a compat layer if there are mixed
serialized
- (BaseSerialisation and serde) data, which can happen if a deferred task
resumes after upgrade.
+ The values are deliberately left encoded. Only the worker resuming the
task needs them as
+ objects; decoding them here would reconstruct task-supplied values in the
scheduler, the
+ triggerer and the API server for nothing, since all the caller does is add
the event.
- The result is checked here rather than assumed, so callers never have to
trust the shape of
- what comes back out of the stored payload.
+ Two encodings are accepted: serde, which stores a dict as a plain dict of
encoded values, and
+ the legacy ``{"__type": "dict", "__var": {...}}`` form of rows written
before the switch to
+ serde, which a deferred task may still carry across an upgrade.
- :raise ValueError: The payload did not decode to a dict.
- :raise Exception: Whatever the two decoders raise on a payload they cannot
read -- the stored
- blob is arbitrary, so the set is open and callers have to treat it as
such.
+ :return: The encoded mapping, and whether it came from the legacy form.
+ :raise ValueError: The stored payload is not an encoded dict.
"""
- from airflow.sdk.serde import deserialize
-
- try:
- next_kwargs = deserialize(next_kwargs_raw)
- except (ImportError, KeyError, AttributeError, TypeError):
- from airflow.serialization.serialized_objects import BaseSerialization
-
- next_kwargs = BaseSerialization.deserialize(next_kwargs_raw)
-
- if not isinstance(next_kwargs, dict):
- raise ValueError(f"next_kwargs decoded to
{type(next_kwargs).__name__}, expected a dict")
- return next_kwargs
+ from airflow._shared.serialization import CLASSNAME, OLD_DATA, OLD_DICT,
OLD_TYPE, SCHEMA_ID
+
+ if not isinstance(next_kwargs_raw, dict):
+ raise ValueError(f"next_kwargs is stored as
{type(next_kwargs_raw).__name__}, expected a dict")
+ if OLD_TYPE in next_kwargs_raw and OLD_DATA in next_kwargs_raw:
+ if next_kwargs_raw[OLD_TYPE] != OLD_DICT or not
isinstance(next_kwargs_raw[OLD_DATA], dict):
+ raise ValueError(
+ f"next_kwargs is stored as a legacy
{next_kwargs_raw[OLD_TYPE]!r}, expected a dict"
+ )
+ return next_kwargs_raw[OLD_DATA], True
+ if CLASSNAME in next_kwargs_raw or SCHEMA_ID in next_kwargs_raw:
+ raise ValueError(
+ f"next_kwargs is stored as an encoded
{next_kwargs_raw.get(CLASSNAME)!r}, expected a dict"
+ )
+ return next_kwargs_raw, False
def _fail_unresumable_task_instance(
@@ -618,7 +622,7 @@ def handle_event_submit(event: TriggerEvent, *,
task_instance: TaskInstance, ses
# stored kwargs for a payload the trigger just yielded would point the
author at DB state that
# was never the problem.
try:
- next_kwargs = _decode_next_kwargs(next_kwargs_raw)
+ stored_kwargs, is_legacy = _split_stored_next_kwargs(next_kwargs_raw)
except Exception as exc:
log.exception(
"Could not decode the stored next_kwargs of %s; failing it instead
of resuming it",
@@ -633,14 +637,20 @@ def handle_event_submit(event: TriggerEvent, *,
task_instance: TaskInstance, ses
)
return
- # Add event to the plain dict, then serialize everything together so nested
- # non-primitive values get proper serde encoding.
- next_kwargs["event"] = event.payload
+ # Encode only the event and add it next to the stored values, keeping the
stored encoding. The
+ # Execution API version converter (ModifyDeferredTaskKwargsToJsonValue)
handles converting
+ # serde to BaseSerialization format when serving old workers.
try:
- # Re-serialize using serde. The Execution API version converter
- # (ModifyDeferredTaskKwargsToJsonValue) handles converting this to
- # BaseSerialization format when serving old workers.
- serialized_next_kwargs = serialize(next_kwargs)
+ if is_legacy:
+ from airflow._shared.serialization import OLD_DATA
+ from airflow.serialization.serialized_objects import
BaseSerialization
+
+ serialized_next_kwargs = {
+ **next_kwargs_raw,
+ OLD_DATA: {**stored_kwargs, "event":
BaseSerialization.serialize(event.payload, strict=True)},
+ }
+ else:
+ serialized_next_kwargs = {**stored_kwargs, "event":
serialize(event.payload)}
except Exception as exc:
log.exception(
"Could not serialize the event payload for %s; failing it instead
of resuming it",
diff --git a/airflow-core/tests/unit/models/test_trigger.py
b/airflow-core/tests/unit/models/test_trigger.py
index d245466a162..3e673126d03 100644
--- a/airflow-core/tests/unit/models/test_trigger.py
+++ b/airflow-core/tests/unit/models/test_trigger.py
@@ -525,6 +525,76 @@ def
test_handle_event_submit_fails_task_with_unusable_next_kwargs(
assert isinstance(task_instance.next_kwargs["traceback"], list)
+def test_handle_event_submit_keeps_stored_values_encoded(session,
create_task_instance, monkeypatch):
+ """
+ Stored values are carried over as encoded, never reconstructed: only the
worker resuming the
+ task needs them as objects, so submitting the event must not run their
constructors.
+ """
+ from airflow.sdk import Asset
+ from airflow.sdk.serde import serialize
+
+ stored = {"asset": serialize(Asset("resume-data")), "count": 1}
+
+ task_instance = create_task_instance(
+ session=session, logical_date=timezone.utcnow(), state=State.DEFERRED
+ )
+ task_instance.next_method = "execute_complete"
+ task_instance.next_kwargs = stored
+ session.flush()
+
+ constructed = []
+ original_init = Asset.__init__
+
+ def spy(self, *args, **kwargs):
+ constructed.append(True)
+ original_init(self, *args, **kwargs)
+
+ monkeypatch.setattr(Asset, "__init__", spy)
+
+ handle_event_submit(TriggerEvent("payload"), task_instance=task_instance,
session=session)
+
+ assert constructed == []
+ session.refresh(task_instance)
+ assert task_instance.state == State.SCHEDULED
+ assert task_instance.next_method == "execute_complete"
+ assert task_instance.next_kwargs == {**stored, "event": "payload"}
+
+
+def test_handle_event_submit_keeps_legacy_encoding(session,
create_task_instance):
+ """Rows still in the legacy ``{"__type": "dict", "__var": ...}`` form get
the event in that form."""
+ task_instance = create_task_instance(
+ session=session, logical_date=timezone.utcnow(), state=State.DEFERRED
+ )
+ task_instance.next_method = "execute_complete"
+ task_instance.next_kwargs = BaseSerialization.serialize({"count": 1})
+ session.flush()
+
+ handle_event_submit(TriggerEvent("payload"), task_instance=task_instance,
session=session)
+
+ session.refresh(task_instance)
+ assert task_instance.state == State.SCHEDULED
+ assert task_instance.next_kwargs == BaseSerialization.serialize({"count":
1, "event": "payload"})
+
+
+def
test_handle_event_submit_fails_task_when_legacy_event_payload_cannot_be_serialized(
+ session, create_task_instance
+):
+ """Legacy-form rows must not resume with an unsupported payload cast to
its ``str()``."""
+ task_instance = create_task_instance(
+ session=session, logical_date=timezone.utcnow(), state=State.DEFERRED
+ )
+ task_instance.next_method = "execute_complete"
+ task_instance.next_kwargs = BaseSerialization.serialize({"count": 1})
+ session.flush()
+
+ handle_event_submit(TriggerEvent({"data": b"abc"}),
task_instance=task_instance, session=session)
+
+ session.refresh(task_instance)
+ assert task_instance.state == State.SCHEDULED
+ assert task_instance.next_method == "__fail__"
+ assert "event payload could not be serialized" in
task_instance.next_kwargs["error"]
+
+
def
test_handle_event_submit_fails_task_when_the_event_payload_cannot_be_serialized(
session, create_task_instance
):
diff --git a/generated/known_sdk_imports_in_core.txt
b/generated/known_sdk_imports_in_core.txt
index 738a73fd880..b0019965223 100644
--- a/generated/known_sdk_imports_in_core.txt
+++ b/generated/known_sdk_imports_in_core.txt
@@ -18,7 +18,7 @@ airflow-core/src/airflow/models/connection.py::5
airflow-core/src/airflow/models/dag.py::1
airflow-core/src/airflow/models/dagrun.py::2
airflow-core/src/airflow/models/taskinstance.py::1
-airflow-core/src/airflow/models/trigger.py::4
+airflow-core/src/airflow/models/trigger.py::3
airflow-core/src/airflow/models/variable.py::8
airflow-core/src/airflow/models/xcom_arg.py::1
airflow-core/src/airflow/plugins_manager.py::1
