amoghrajesh commented on code in PR #70685:
URL: https://github.com/apache/airflow/pull/70685#discussion_r3710011312
##########
airflow-core/src/airflow/models/trigger.py:
##########
@@ -500,6 +500,60 @@ 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.
+ :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.
+ """
+ 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, exc: BaseException, *, 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.
+
+ The traceback travels in ``next_kwargs`` because that is the only channel
that reaches the
+ task log; the exception is otherwise only logged where this runs, which
the Dag author may
+ have no access to. It has to stay the list ``format_exception`` returns --
the runtime joins it.
+ """
Review Comment:
```suggestion
"""
Route through ``__fail__`` (mirrors `Trigger.submit_failure`) so a
worker fails the
task normally instead of leaving it stranded with no event left to
resume it.
Traceback goes into ``next_kwargs`` as a list -- the only channel
reaching the task log --
since ``format_exception`` returns it that way and the runtime joins it.
"""
```
--
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]
