potiuk commented on code in PR #66002:
URL: https://github.com/apache/airflow/pull/66002#discussion_r3173100034
##########
airflow-core/src/airflow/models/taskinstance.py:
##########
@@ -1633,7 +1633,19 @@ def defer_task(self, session: Session = NEW_SESSION) ->
bool:
assert isinstance(self.task, Operator)
if start_trigger_args := self.start_trigger_args:
- trigger_kwargs = start_trigger_args.trigger_kwargs or {}
+ from airflow.serialization.enums import Encoding
+
+ def _normalize(d: Any) -> Any:
+ # trigger_kwargs arrives with Encoding enum keys from
BaseSerialization.
+ # On Python 3.10, str(Encoding.TYPE) returns "Encoding.TYPE"
not "__type",
+ # so we convert enum keys to their values before passing to
serde.serialize.
+ if isinstance(d, dict):
+ return {
+ (k.value if isinstance(k, Encoding) else str(k)):
_normalize(v) for k, v in d.items()
+ }
+ return d
+
+ trigger_kwargs = _normalize(start_trigger_args.trigger_kwargs or
{})
Review Comment:
(About `self.next_kwargs = start_trigger_args.next_kwargs or {}` a few lines
below — line 1661 in the new file.)
After this PR `start_trigger_args.next_kwargs` is also raw
`BaseSerialization`-encoded form (`{Encoding.TYPE: "dict", Encoding.VAR:
{...}}`), but only `trigger_kwargs` gets `_normalize`d above. That line writes
the encoded form (with `Encoding` enum keys) into the `next_kwargs` JSON column.
`_handle_submit_event` in `models/trigger.py` reads it back via
`serde.deserialize` with a `BaseSerialization.deserialize` fallback, and
`serde._convert` does happen to handle the `OLD_TYPE`/`OLD_DATA` shape — so it
might work on 3.11+. On 3.10 the same `Encoding.TYPE` → `"Encoding.TYPE"` key
mangling will hit through `json.dumps` of the JSON column, and neither
`_convert` nor the fallback will recognize the mangled keys.
There is no test coverage for the case where
`start_trigger_args.next_kwargs` is non-`None` (the new test in
`test_dag_serialization.py` uses `next_kwargs=None`). Either:
- Apply the same normalization to `next_kwargs` symmetrically, or
- If next_kwargs really isn't carried through start_from_trigger today,
leave it as `None` rather than persisting the encoded form
Worth adding an end-to-end test that defers a task with non-`None`
`next_kwargs`, persists, resumes, and asserts the resumed kwargs match the
original.
---
Drafted-by: Claude Opus 4.7; reviewed by @potiuk before posting
##########
airflow-core/src/airflow/models/taskinstance.py:
##########
@@ -1633,7 +1633,19 @@ def defer_task(self, session: Session = NEW_SESSION) ->
bool:
assert isinstance(self.task, Operator)
if start_trigger_args := self.start_trigger_args:
- trigger_kwargs = start_trigger_args.trigger_kwargs or {}
+ from airflow.serialization.enums import Encoding
+
+ def _normalize(d: Any) -> Any:
+ # trigger_kwargs arrives with Encoding enum keys from
BaseSerialization.
+ # On Python 3.10, str(Encoding.TYPE) returns "Encoding.TYPE"
not "__type",
Review Comment:
The "On Python 3.10, ..." framing reads as if 3.10 is an edge case, but
`DEFAULT_PYTHON_MAJOR_MINOR_VERSION` in
`dev/breeze/src/airflow_breeze/global_constants.py` is still `"3.10"` — it's
the default CI target, not an outlier. Suggest rephrasing as "needed for Python
3.10 compatibility: `str(Encoding.TYPE)` returns 'Encoding.TYPE' there, not
'__type'", so it doesn't read as if 3.10 is unusual.
---
Drafted-by: Claude Opus 4.7; reviewed by @potiuk before posting
##########
airflow-core/tests/unit/serialization/test_dag_serialization.py:
##########
@@ -2627,6 +2643,38 @@ def execute_complete(self):
}
assert tasks[1]["__var"]["start_from_trigger"] is True
+ def test_trigger_kwargs_not_deserialised_through_serdag(self):
Review Comment:
Nice regression guard for the raw-JSON contract. Two coverage gaps worth
filling while you're here:
1. A defer/resume test on Python 3.10 specifically (parametrize via
`sys.version_info` skip if needed) that exercises the `_normalize` path
end-to-end, so the next time someone moves it the test fails at the right place.
2. A test where `start_trigger_args.next_kwargs` is non-`None` (with a value
that contains a `timedelta` or other type that BaseSerialization encodes),
defers, persists, resumes, and asserts the value round-trips. Right now
`next_kwargs` carries the encoded form past `defer_task` with no test to catch
it if it breaks.
---
Drafted-by: Claude Opus 4.7; reviewed by @potiuk before posting
##########
airflow-core/tests/unit/serialization/test_dag_serialization.py:
##########
@@ -855,6 +857,20 @@ def validate_deserialized_task(
else:
assert serialized_task.resources == task.resources
+ # start_trigger_args: trigger_kwargs is kept as raw JSON after
deserialization;
+ # compare after deserializing both sides
+ if task.start_trigger_args is not None:
+ from airflow.serialization.serialized_objects import
BaseSerialization
+
+ s = serialized_task.start_trigger_args
+ o = task.start_trigger_args
+ assert s.trigger_cls == o.trigger_cls
+ assert s.next_method == o.next_method
Review Comment:
This block deserializes both sides through `BaseSerialization.deserialize`
and compares, which papers over the contract change in tests. The CI failure
(`test_roundtrip_provider_example_dags` → `KeyError: <Encoding.VAR: '__var'>`)
is exactly this code tripping over `Encoding`-key vs string-key shape after a
JSON round-trip on 3.10.
Two cleaner options:
1. Compare raw encoded forms directly (matching what's actually stored), and
add a separate, explicit test that round-trips a single trigger through
`BaseSerialization.deserialize`.
2. Apply the same `_stringify_encoding_keys` helper in this comparison so
both sides go through the same normalization that the production code uses.
Either way, the comparison should not silently rely on a code path that's
known to be flaky across Python versions.
---
Drafted-by: Claude Opus 4.7; reviewed by @potiuk before posting
##########
airflow-core/src/airflow/models/taskinstance.py:
##########
@@ -1633,7 +1633,19 @@ def defer_task(self, session: Session = NEW_SESSION) ->
bool:
assert isinstance(self.task, Operator)
if start_trigger_args := self.start_trigger_args:
- trigger_kwargs = start_trigger_args.trigger_kwargs or {}
+ from airflow.serialization.enums import Encoding
+
+ def _normalize(d: Any) -> Any:
Review Comment:
This `_normalize` is a band-aid for one consumer (`defer_task`) and doesn't
help anywhere else that touches `start_trigger_args.trigger_kwargs` after this
PR. Two of the failing CI tests on Python 3.10 illustrate that:
- `test_trigger_kwargs_serialization_cleanup` constructs
`Trigger(kwargs=BaseSerialization.serialize(kw))` directly — bypassing
`defer_task` — and the `Encoding`-keyed dict reaches `serde.serialize`
unfiltered. On 3.10 this produces `"Encoding.TYPE"` keys in the encrypted blob,
which is exactly the bug the test is meant to catch.
- `test_render_template_fields` reads `trigger.template_fields` and finds
the encoded `{"__type": "tuple", "__var": []}` form instead of a real tuple.
Pulling this normalization down into `Trigger.encrypt_kwargs` (or applying
it inside `_decode_start_trigger_args` on the way out) would cover all
consumers in one place. Defining `_normalize` as a closure inside `defer_task`
also makes it harder to test on its own — worth promoting to a module-level
helper named e.g. `_stringify_encoding_keys` regardless of where it ends up
being called.
One more thing: `_normalize` only recurses into `dict`.
`BaseSerialization.serialize` does encode tuples / lists with `Encoding.TYPE:
"tuple"` shape, and those can be nested under a list-of-dicts. Worth handling
lists/tuples in the recursion too.
---
Drafted-by: Claude Opus 4.7; reviewed by @potiuk before posting
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
