potiuk commented on code in PR #70685:
URL: https://github.com/apache/airflow/pull/70685#discussion_r3708373090
##########
airflow-core/src/airflow/models/trigger.py:
##########
@@ -509,33 +554,39 @@ def handle_event_submit(event: TriggerEvent, *,
task_instance: TaskInstance, ses
as well as its state to scheduled. It also adds the event's payload
into the kwargs for the task.
+ A task instance whose stored kwargs cannot be decoded (or re-encoded with
the payload added)
+ is failed rather than resumed. That work happens in the scheduler and API
processes, which
+ handle every parked task instance in turn, so a single unusable payload
must not be able to
+ abort the caller.
+
:param task_instance: The task instance to handle the submit event for.
:param session: The session to be used for the database callback sink.
"""
- from airflow.sdk.serde import deserialize, serialize
+ from airflow.sdk.serde import serialize
from airflow.utils.state import TaskInstanceState
next_kwargs_raw = task_instance.next_kwargs or {}
- # deserialize 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
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)
-
- # Add event to the plain dict, then serialize everything together so nested
- # non-primitive values get proper serde encoding.
- if TYPE_CHECKING:
- assert isinstance(next_kwargs, dict)
- next_kwargs["event"] = event.payload
+ next_kwargs = _decode_next_kwargs(next_kwargs_raw)
+ # Add event to the plain dict, then serialize everything together so
nested
+ # non-primitive values get proper serde encoding.
+ next_kwargs["event"] = event.payload
+ # 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)
Review Comment:
This is the sharpest of the batch, and I want to be straight about it rather
than wave it through: for that specific case the change does trade one bad
outcome for a different bad outcome. A HITL response whose `params_input`
carries `__classname__` or `__schema_id__` currently 500s and can be corrected
and resubmitted; after this change it is recorded, the task is routed to
`__fail__`, the route answers 200, and the `response_received` guard makes the
corrected resubmission a 409. That is worse for that user, and it is caused by
this PR.
I am still landing it as is, because the alternative it replaces is worse in
aggregate: the same raise in the triggerer leaves the event unconfirmed and
redelivers it forever, and in the sweep it aborts the whole batch, taking
unrelated healthy task instances with it. Trading "one user must contact an
admin" for "one payload wedges a component" is the right way round. The
reserved-key case also needs a params key literally named `__classname__` or
`__schema_id__`, which is narrow.
The write-side 400 in `hitl.py` is the right fix and I agree it belongs
there rather than here — validating what is accepted, not compensating after
storing it.
On whether this function should report its outcome: it now does, without a
signature change. `_fail_unresumable_task_instance` sets `next_method` to
`TRIGGER_FAIL_REPR`, so a caller can check `ti.next_method ==
TRIGGER_FAIL_REPR` immediately after the call — which is exactly what the sweep
now does for its summary counts. The HITL route could use the same check to
answer something other than 200. I have not made that change here because it
alters a public endpoint's response, which is not something I want to slip into
a 3.3.1 backport, but the mechanism is in place for it.
---
Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting
##########
airflow-core/src/airflow/models/trigger.py:
##########
@@ -500,6 +500,51 @@ def get_sorted_triggers(
return result
+def _decode_next_kwargs(next_kwargs_raw: Any) -> dict[str, Any]:
+ """
+ Decode the stored ``next_kwargs`` of a task instance into a plain dict.
+
+ 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 result is checked here rather than assumed, so callers never have to
trust the shape of
+ what comes back out of the stored payload.
+
+ :raise ValueError: The payload did not decode to a 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
+
+
+def _fail_unresumable_task_instance(task_instance: TaskInstance, reason: str,
*, session: Session) -> None:
+ """
+ Re-queue a task instance that cannot be resumed, so that a worker fails it.
+
+ Mirrors :meth:`Trigger.submit_failure`: the special ``__fail__``
next_method makes the worker
+ fail the task immediately, which runs its normal failure handling (retries
and callbacks
+ included). Leaving the task instance parked instead would strand it there,
as the event that
+ should have resumed it is already gone.
+ """
+ task_instance.next_method = TRIGGER_FAIL_REPR
+ task_instance.next_kwargs = {"error": reason}
+ # Remove ourselves as its trigger
Review Comment:
Removed — both restated the line directly beneath them.
---
Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting
##########
airflow-core/src/airflow/models/trigger.py:
##########
@@ -500,6 +500,51 @@ def get_sorted_triggers(
return result
+def _decode_next_kwargs(next_kwargs_raw: Any) -> dict[str, Any]:
+ """
+ Decode the stored ``next_kwargs`` of a task instance into a plain dict.
+
+ 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 result is checked here rather than assumed, so callers never have to
trust the shape of
+ what comes back out of the stored payload.
+
+ :raise ValueError: The payload did not decode to a 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
+
+
+def _fail_unresumable_task_instance(task_instance: TaskInstance, reason: str,
*, session: Session) -> None:
+ """
+ Re-queue a task instance that cannot be resumed, so that a worker fails it.
+
+ Mirrors :meth:`Trigger.submit_failure`: the special ``__fail__``
next_method makes the worker
+ fail the task immediately, which runs its normal failure handling (retries
and callbacks
+ included). Leaving the task instance parked instead would strand it there,
as the event that
+ should have resumed it is already gone.
+ """
+ task_instance.next_method = TRIGGER_FAIL_REPR
+ task_instance.next_kwargs = {"error": reason}
+ # Remove ourselves as its trigger
+ task_instance.trigger_id = None
+ # Finally, mark it as scheduled so it gets re-queued
Review Comment:
Removed — both restated the line directly beneath them.
---
Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting
##########
airflow-core/tests/unit/jobs/test_scheduler_job.py:
##########
@@ -8077,6 +8077,60 @@ def
test_awaiting_input_timeout_without_defaults_fails(self, dag_maker):
assert ti.next_method == "execute_complete"
assert ti.next_kwargs["event"]["error_type"] == "timeout"
+ def test_awaiting_input_timeout_sweep_survives_unusable_next_kwargs(self,
dag_maker):
Review Comment:
Trimmed — the docstring keeps the intent in one line and the inline
narration is gone.
---
Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting
##########
airflow-core/tests/unit/models/test_trigger.py:
##########
@@ -256,6 +257,45 @@ def test_submit_event_no_n_plus_one_for_assets(_, session,
asset_count, expected
Trigger.submit_event(trigger_id, TriggerEvent("payload"),
session=session)
[email protected](
+ "stored_next_kwargs",
+ [
+ # Decoding blows up: serde rejects the class name, and the
BaseSerialization fallback then
+ # trips over the missing legacy keys.
+ pytest.param(
+ {"__classname__": "not.allowed.Thing", "__version__": 1,
"__data__": {}},
+ id="undecodable",
+ ),
+ # Decodes cleanly, but not into a dict: legacy encoding of a bare
datetime.
+ pytest.param({"__type": "datetime", "__var": 1735689600.0},
id="not-a-dict"),
+ ],
+)
+def test_handle_event_submit_fails_task_with_unusable_next_kwargs(
Review Comment:
Trimmed — the docstring keeps the intent in one line and the inline
narration is gone.
---
Drafted-by: Claude Code (Opus 5); 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]
