amoghrajesh commented on code in PR #70685:
URL: https://github.com/apache/airflow/pull/70685#discussion_r3701562717
##########
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:
```suggestion
```
##########
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:
Docstring and inline comments are narrating same thing, remove either.
##########
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:
Same comment as above here.
##########
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:
```suggestion
```
--
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]
