ephraimbuddy commented on code in PR #72721:
URL: https://github.com/apache/airflow/pull/72721#discussion_r4006472504
##########
airflow-core/src/airflow/models/dagrun.py:
##########
@@ -1511,30 +1517,66 @@ def notify_dagrun_state_changed(self, msg: str):
# we can't get all the state changes on SchedulerJob,
# or LocalTaskJob, so we don't want to "falsely advertise" we notify
about that
+ def _build_callback_last_ti(self, relevant_ti: TI, *, session: Session) ->
TIDataModel | None:
+ """
+ Build a callback context's ``last_ti``, standing in a Dag version if
the record has none.
+
+ ``session`` is required, not defaulted: ``settings.Session`` is a
``scoped_session``,
+ so acquiring one here would hand back the caller's own and then commit
and close it.
+ """
+ from airflow.api_fastapi.execution_api.datamodels.taskinstance import
TaskInstance as TIDataModel
+ from airflow.models.dag_version import DagVersion
+
+ if relevant_ti.dag_version_id is not None:
+ return TIDataModel.model_validate(relevant_ti,
from_attributes=True)
+
+ dag_version_id = self.created_dag_version_id
Review Comment:
Added the explanation of the difference between this and the one in
scheduler. the fallback only populates the callback payload; row backfilling
stays with the scheduler. If the scheduler has already filled the version, we
preserve 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]
