kaxil commented on code in PR #66608:
URL: https://github.com/apache/airflow/pull/66608#discussion_r3345614622
##########
airflow-core/src/airflow/triggers/callback.py:
##########
@@ -41,6 +41,9 @@ def __init__(self, callback_path: str, callback_kwargs:
dict[str, Any] | None =
super().__init__()
self.callback_path = callback_path
self.callback_kwargs = callback_kwargs or {}
+ # Context is set by the TriggerRunner (from dag_run_data) before run()
is called.
+ # This attribute is set externally; defaults to None for triggers
without context.
+ self.context: dict | None = None
Review Comment:
The reply says renamed to `self._callback_context`, but at HEAD (`fa98a6d4`)
it's still `self.context`: the attribute (`callback.py:91`), the read in
`run()` (`callback.py:107`), and the assignment in the triggerer
(`triggerer_job_runner.py:1318`). `_callback_context` doesn't appear anywhere
in the tree. The docstring note did land, so only the rename is outstanding,
and the shadowing / not-serialized concern from the original comment still
stands.
##########
airflow-core/src/airflow/models/deadline.py:
##########
@@ -215,46 +216,32 @@ def prune_deadlines(cls, *, session: Session, conditions:
dict[Mapped, Any]) ->
def handle_miss(self, session: Session):
"""Handle a missed deadline by queueing the callback."""
-
- def get_simple_context():
- from airflow.api_fastapi.core_api.datamodels.dag_run import
DAGRunResponse
- from airflow.models import DagRun
-
- # TODO: Use the TaskAPI from within Triggerer to fetch full
context instead of sending this context
- # from the scheduler
-
- # Fetch the DagRun from the database again to avoid errors when
self.dagrun's relationship fields
- # are not in the current session.
- dagrun = session.get(DagRun, self.dagrun_id)
-
- return {
- "dag_run":
DAGRunResponse.model_validate(dagrun).model_dump(mode="json"),
- "deadline": {"id": self.id, "deadline_time":
self.deadline_time},
- }
+ # Routing identifiers: stored at the top level of callback.data so the
triggerer/executor
+ # can locate the DagRun and build execution context *before* invoking
the callback.
+ # These are NOT user payload — they are consumed by infrastructure
(_fetch_callback_dag_run_data
+ # in triggerer_job_runner and _fetch_and_build_context in
callback_supervisor) and stripped
+ # before the callback function is called.
+ # TODO: A dedicated `routing_data` field on Callback would make this
separation explicit
+ # and avoid mixing routing metadata with user kwargs in the same JSON
blob. Deferred because
+ # changing the Callback model would require a migration and
coordination with the executor path.
+ self.callback.data["dag_id"] = self.dagrun.dag_id
+ self.callback.data["run_id"] = self.dagrun.run_id
+
+ # Deadline-specific info goes into kwargs so it's passed to the
callback function.
+ self.callback.data["kwargs"]["deadline_id"] = str(self.id)
Review Comment:
The kwargs `TypeError` is fixed: `deadline_id`/`deadline_time` now go to the
top level of `callback.data` rather than `data["kwargs"]`, so they're no longer
passed to the user callable, and the executor path re-exposes them as
`context["deadline"]` (`callback_supervisor.py:244-245`). But this comment also
credits `_fetch_callback_dag_run_data` (the triggerer path) with exposing them,
and it doesn't: it returns only `dagrun.dag_run_data`
(`triggerer_job_runner.py:840`), and `build_context_from_dag_run` never adds a
`deadline` key, so `context["deadline"]` is absent for `TriggererCallback`. A
deadline that routes async and templates `{{ deadline.deadline_time }}`, the
exact pattern in `deadline-alerts.rst`, renders empty. Either inject the
deadline dict on the triggerer side too, or scope this comment to the executor
path.
--
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]
