seanghaeli commented on code in PR #68917:
URL: https://github.com/apache/airflow/pull/68917#discussion_r3531936264
##########
airflow-core/src/airflow/serialization/definitions/dag.py:
##########
@@ -715,51 +715,125 @@ def _process_dagrun_deadline_alerts(
if not deadline_alert:
continue
- deserialized_deadline_alert = decode_deadline_alert(
- {
- Encoding.TYPE: DAT.DEADLINE_ALERT,
- Encoding.VAR: {
- DeadlineAlertFields.REFERENCE:
deadline_alert.reference,
- DeadlineAlertFields.INTERVAL: deadline_alert.interval,
- DeadlineAlertFields.CALLBACK:
deadline_alert.callback_def,
- },
- }
- )
+ # Isolate each deadline alert: creating a deadline is auxiliary to
creating the
+ # DagRun itself, and must never prevent the DagRun from being
created. A single bad
+ # alert -- e.g. a ``VariableInterval`` whose backing Variable is
missing / non-integer
+ # / <= 0 (``coerce_to_timedelta`` raises ``ValueError``), or a
reference whose
+ # ``evaluate_with`` fails -- would otherwise propagate out of
``create_dagrun`` and
+ # abort the whole run, silently stopping the DAG from scheduling.
+ #
+ # Isolation is done with a plain ``try``/``except`` and MUST NOT
use
+ # ``session.begin_nested()``: ``create_dagrun`` runs on the
scheduler session inside
+ # the ``prohibit_commit`` guard, and releasing a SAVEPOINT issues
a commit, which trips
+ # that guard (``RuntimeError("UNEXPECTED COMMIT ...")``) and --
because this very block
+ # swallows it -- silently skips deadline creation for *every*
scheduled DagRun. The
+ # try/except alone is sufficient: the only DB mutation in the loop
body is the final
+ # ``session.add`` (everything before it is a decode, an in-memory
resolution, or a
+ # read-only ``evaluate_with`` query), so an exception leaves no
partial state to undo,
+ # and the pending ``Deadline`` is persisted by the caller's outer
transaction.
+ try:
+ deserialized_deadline_alert = decode_deadline_alert(
+ {
+ Encoding.TYPE: DAT.DEADLINE_ALERT,
+ Encoding.VAR: {
+ DeadlineAlertFields.REFERENCE:
deadline_alert.reference,
+ DeadlineAlertFields.INTERVAL:
deadline_alert.interval,
+ DeadlineAlertFields.CALLBACK:
deadline_alert.callback_def,
+ },
+ }
+ )
- interval = deserialized_deadline_alert.interval
+ interval = deserialized_deadline_alert.interval
- if isinstance(interval, VariableInterval):
- interval = interval.resolve()
+ if isinstance(interval, VariableInterval):
+ interval = self._resolve_variable_interval(interval,
session=session)
- if isinstance(deserialized_deadline_alert.reference,
SerializedReferenceModels.TYPES.DAGRUN):
- deadline_time =
deserialized_deadline_alert.reference.evaluate_with(
- session=session,
- interval=interval,
- # TODO : Pretty sure we can drop these last two; verify
after testing is complete
- dag_id=self.dag_id,
- run_id=orm_dagrun.run_id,
- )
+ if isinstance(deserialized_deadline_alert.reference,
SerializedReferenceModels.TYPES.DAGRUN):
+ deadline_time =
deserialized_deadline_alert.reference.evaluate_with(
+ session=session,
+ interval=interval,
+ # TODO : Pretty sure we can drop these last two;
verify after testing is complete
+ dag_id=self.dag_id,
+ run_id=orm_dagrun.run_id,
+ )
- if deadline_time is not None:
- session.add(
- Deadline(
- deadline_time=deadline_time,
- callback=deserialized_deadline_alert.callback,
- dagrun_id=orm_dagrun.id,
- deadline_alert_id=deadline_alert.id,
- dag_id=orm_dagrun.dag_id,
- bundle_name=orm_dagrun.dag_model.bundle_name,
+ if deadline_time is not None:
+ session.add(
+ Deadline(
+ deadline_time=deadline_time,
+ callback=deserialized_deadline_alert.callback,
+ dagrun_id=orm_dagrun.id,
+ deadline_alert_id=deadline_alert.id,
+ dag_id=orm_dagrun.dag_id,
+ bundle_name=orm_dagrun.dag_model.bundle_name,
+ )
)
- )
- team_name = (
- DagModel.get_team_name(self.dag_id, session=session)
- if airflow_conf.getboolean("core", "multi_team")
- else None
- )
- stats.incr(
- "deadline_alerts.deadline_created",
- tags=prune_dict({"dag_id": self.dag_id, "team_name":
team_name}),
- )
+ team_name = (
+ DagModel.get_team_name(self.dag_id,
session=session)
+ if airflow_conf.getboolean("core", "multi_team")
+ else None
+ )
+ stats.incr(
+ "deadline_alerts.deadline_created",
+ tags=prune_dict({"dag_id": self.dag_id,
"team_name": team_name}),
+ )
+ except Exception:
Review Comment:
Anything escaping would stop the DagRun being created, so I'd rather except
broadly. There could be other failure modes so what makes sense to me is to
except broadly and log the traceback.
--
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]
