kaxil commented on code in PR #75765:
URL: https://github.com/apache/airflow/pull/75765#discussion_r4390070075
##########
airflow-core/src/airflow/serialization/definitions/dag.py:
##########
@@ -160,6 +168,23 @@ def __init__(self, *, dag_id: str) -> None:
def __repr__(self) -> str:
return f"<SerializedDAG: {self.dag_id}>"
+ @property
+ def timetable(self) -> Timetable:
Review Comment:
Deferring the decode moves a timetable failure past the scheduler's load
guards. `_start_queued_dagruns` only wraps `cached_get_dag(dag_run)` (the guard
#74499 added), so a queued run whose timetable no longer decodes now raises
`TimetableNotRegistered` from `dag.timetable.periodic` in `_update_state`,
outside any try, and the scheduler exits. The run is still queued, so every
restart hits it again. I reproduced this in breeze with a queued run of a
`CustomSerializationTimetable` Dag and the plugin unregistered; with the base
versions of `dag.py` and `serialized_objects.py` the bad run is skipped and the
good one starts. The existing test for that guard mocks `get_dag_for_run` to
raise, so it never reaches the lazy decode.
Same shape at `timetable = dag.timetable` in
`_create_dagruns_for_partitioned_asset_dags`, and after `_get_current_dag`
returns in `_create_dag_runs` (`calculate_dagrun_date_fields` in the
run-already-exists branch, `_collect_gated_asset_events`) and
`_create_dag_runs_asset_triggered` (`dag.timetable.asset_triggered`). Could the
scheduler read `.timetable` inside those existing try blocks, or make the lazy
decode opt-in for the Execution API's dag bag only? Either way a scheduler test
that stores a real undecodable timetable would pin it.
##########
airflow-core/src/airflow/serialization/serialized_objects.py:
##########
@@ -1223,8 +1210,8 @@ def populate_operator(
elif k == "_on_failure_fail_dagrun":
k = "on_failure_fail_dagrun"
elif k == "weight_rule":
+ # Decoded by SerializedBaseOperator.weight_rule, so loading a
Dag does not need the plugin.
k = "_weight_rule"
Review Comment:
This has the same effect on the scheduler for `weight_rule`. A Dag whose
custom priority strategy isn't registered on the scheduler used to fail inside
`_get_current_dag` and get skipped. Now it loads, and `create_dagrun` ->
`verify_integrity` -> `TI.insert_mapping` reads `task.weight_rule` and raises
`ValueError`. `_create_dag_runs` catches that, but
`_create_dag_runs_asset_triggered` has no try around `dag.create_dagrun`, so an
asset-triggered Dag with that strategy would take the loop down. I traced this
one rather than running it.
##########
airflow-core/src/airflow/serialization/definitions/dag.py:
##########
@@ -160,6 +168,23 @@ def __init__(self, *, dag_id: str) -> None:
def __repr__(self) -> str:
return f"<SerializedDAG: {self.dag_id}>"
+ @property
+ def timetable(self) -> Timetable:
+ if (encoded := self._encoded_timetable) is not None:
+ try:
+ self._timetable = decode_timetable(encoded)
+ except TimetableNotRegistered:
+ raise
+ except Exception as err:
+ raise DeserializationError(self.dag_id) from err
Review Comment:
Nothing exercises this wrap or the decode-once caching. A bare `raise` here,
or dropping `self._encoded_timetable = None` below, would both leave the suite
green (`validate_serialized_dag` unwraps `__cause__`, and the timetable
assertions compare by equality). A `CustomSerializationTimetable` payload with
`__var: {}` makes `deserialize` raise `KeyError`, which would cover the wrap,
plus a `dag.timetable is dag.timetable` check.
--
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]
