uranusjr commented on code in PR #51949:
URL: https://github.com/apache/airflow/pull/51949#discussion_r2162965905
##########
airflow-core/src/airflow/models/dagrun.py:
##########
@@ -1352,21 +1352,48 @@ def notify_dagrun_state_changed(self, msg: str = ""):
def handle_dag_callback(self, dag: SDKDAG, success: bool = True, reason:
str = "success"):
"""Only needed for `dag.test` where `execute_callbacks=True` is passed
to `update_state`."""
+ task_instances = self.get_task_instances()
+
+ # Identify the most relevant task instance
+ last_relevant_ti = None
+ if not success:
+ failed_tis = [ti for ti in task_instances if ti.state in
State.failed_states and ti.end_date]
+ failed_tis.sort(key=lambda x: x.end_date, reverse=True)
+ last_relevant_ti = failed_tis[0] if failed_tis else None
+ else:
+ success_tis = [ti for ti in task_instances if ti.state in
State.success_states and ti.end_date]
+ success_tis.sort(key=lambda x: x.end_date, reverse=True)
+ last_relevant_ti = success_tis[0] if success_tis else None
Review Comment:
This can be something like
```python
last_relevant_ti = max(success_tis, ...) if success_tis else None
```
instead.
I kind of wonder if we can even avoid building the list at all.
Also is simply sorting by end_date correct? Especially with trigger_rule,
the last success/failed ti might not necessarily be the ti that causes the dag
run to be marked as success/failed. Can you check the logic in 2.x to see how
the ti is selected?
--
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]
