potiuk commented on code in PR #70685:
URL: https://github.com/apache/airflow/pull/70685#discussion_r3708366093
##########
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}
Review Comment:
Done. `_fail_unresumable_task_instance` now takes the exception and packs
`format_exception(type(exc), exc, exc.__traceback__)` into `next_kwargs`,
matching `submit_failure`. Kept as the list `format_exception` returns, since
`operator.py` does `"\n".join(traceback)` when it renders it.
Your framing of why is the part worth keeping, so I put it in the docstring:
`next_kwargs` is the only channel that reaches the task log, and the process
log where this runs is frequently not the Dag author's to read.
---
Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting
##########
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)
+ except Exception as exc:
+ log.exception("Could not process next_kwargs of %s; failing it instead
of resuming it", task_instance)
+ _fail_unresumable_task_instance(
+ task_instance,
+ "Could not resume the task: its stored next_kwargs could not be
processed "
Review Comment:
Agreed, and split. Decoding and re-encoding are now separate `try` blocks
with their own messages — "its stored next_kwargs could not be decoded" versus
"the event payload could not be serialized" — so a payload the trigger just
yielded is no longer reported as unreadable database state.
Added
`test_handle_event_submit_fails_task_when_the_event_payload_cannot_be_serialized`,
which uses the reserved-key case from your other comment to drive the second
branch.
---
Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting
##########
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
Review Comment:
Both taken. The docstring now says "the triggerer, the scheduler and the API
processes" and calls out that the triggerer had the worst of it —
`handle_events` leaving the event unconfirmed meant the poison event came back
indefinitely, which is a materially worse failure than one aborted sweep pass.
Also added the note to `_decode_next_kwargs`: it documents that the
decoders' own failures propagate too and that the set is open, which is exactly
why the caller's `except` is broad. That was the reader-narrows-it-back-down
risk you were pointing at.
---
Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting
##########
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
Review Comment:
Dropped — `TaskInstanceState` is imported at module level on line 43 and
`_fail_unresumable_task_instance` was already relying on that, so the local
import was pure shadowing.
---
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):
+ """
+ A parked task whose stored ``next_kwargs`` cannot be turned into a
dict is failed by the
+ sweep instead of aborting it, and the other timed-out tasks in the
same batch are still
+ resolved. The sweep runs on a recurring scheduler timer, so it has to
finish the batch.
+ """
+ session = settings.Session()
+ with dag_maker(
+ dag_id="test_awaiting_input_bad_kwargs",
+ start_date=DEFAULT_DATE,
+ schedule="@once",
+ session=session,
+ ):
+ EmptyOperator(task_id="dummy1")
+ dr_bad = dag_maker.create_dagrun()
+ dr_good = dag_maker.create_dagrun(
+ run_id="good", logical_date=DEFAULT_DATE +
datetime.timedelta(seconds=1)
+ )
+ ti_bad = dr_bad.get_task_instance("dummy1", session=session)
+ ti_good = dr_good.get_task_instance("dummy1", session=session)
+ for ti in (ti_bad, ti_good):
+ ti.state = State.AWAITING_INPUT
+ ti.trigger_timeout = timezone.utcnow() -
datetime.timedelta(seconds=60)
+ ti.next_method = "execute_complete"
+ ti.next_kwargs = {}
+ # Stored kwargs that no longer decode into a dict (here: a class name
that is not allowed
+ # for deserialization, which the legacy fallback cannot read either).
+ ti_bad.next_kwargs = {"__classname__": "not.allowed.Thing",
"__version__": 1, "__data__": {}}
+ session.add(
+ HITLDetail(
+ ti_id=ti_good.id,
+ options=["Approve", "Reject"],
+ subject="approve?",
+ defaults=["Approve"],
+ multiple=False,
+ params={},
+ )
+ )
+ session.flush()
+
+ self.job_runner = SchedulerJobRunner(job=Job())
+ self.job_runner.check_awaiting_input_timeouts(session=session)
Review Comment:
Fixed here rather than left outside the diff — the log line only becomes
untrue *because* of this change, so it belongs with it.
The sweep now tracks `num_unresumable` separately: after each
`handle_event_submit` it checks `ti.next_method == TRIGGER_FAIL_REPR`, which is
reachable in all three branches and means the task was neither resolved nor
timed out. The summary reads "%i resolved (response/defaults), %i failed, %i
could not be resumed".
`test_awaiting_input_timeout_sweep_survives_unusable_next_kwargs` now asserts
the counts are `(1, 0, 1)` rather than the previous `(1, 0)` with the bad one
hidden inside "resolved".
---
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]
