ashb commented on code in PR #22679:
URL: https://github.com/apache/airflow/pull/22679#discussion_r843729576
##########
airflow/models/dagrun.py:
##########
@@ -810,20 +822,27 @@ def verify_integrity(self, session: Session =
NEW_SESSION):
task = None
try:
task = dag.get_task(ti.task_id)
+
+ should_restore_task = (task is not None) and ti.state ==
State.REMOVED
+ if should_restore_task:
+ self.log.info("Restoring task '%s' which was previously
removed from DAG '%s'", ti, dag)
+ Stats.incr(f"task_restored_to_dag.{dag.dag_id}", 1, 1)
+ ti.state = State.NONE
except AirflowException:
if ti.state == State.REMOVED:
pass # ti has already been removed, just ignore it
elif self.state != State.RUNNING and not dag.partial:
self.log.warning("Failed to get task '%s' for dag '%s'.
Marking it as removed.", ti, dag)
Stats.incr(f"task_removed_from_dag.{dag.dag_id}", 1, 1)
ti.state = State.REMOVED
+ continue
- should_restore_task = (task is not None) and ti.state ==
State.REMOVED
- if should_restore_task:
- self.log.info("Restoring task '%s' which was previously
removed from DAG '%s'", ti, dag)
- Stats.incr(f"task_restored_to_dag.{dag.dag_id}", 1, 1)
- ti.state = State.NONE
- session.merge(ti)
Review Comment:
Yes -- the TIs are fetched from the current attached session so this
shouldn't be required?
--
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]
