jason810496 commented on code in PR #66854:
URL: https://github.com/apache/airflow/pull/66854#discussion_r3622049724
##########
airflow-core/src/airflow/models/taskinstance.py:
##########
@@ -1035,6 +1035,9 @@ def prepare_db_for_next_try(self, session: Session):
TaskInstanceHistory.record_ti(self, session=session)
session.execute(delete(TaskReschedule).filter_by(ti_id=self.id))
+ # Reassigning the id re-points any not-yet-drained asset_event_queue
rows via the FK's
+ # ON UPDATE CASCADE, so the scheduler drain still finds this TI
instead of dropping the
+ # pending asset events.
Review Comment:
After revisiting this part. I think we should not switch to the new
generated TI.id .
We should use the old TI.id instead. Since the AssetEvent table use
source_task_id without FK, if we re-map to a new generated TI.id, we change the
write sementic. IIUC, we should drop the FK of AssetEventQueue with TI.id and
check if the TI and TIH existed in the scheduler.
--
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]
