uranusjr commented on code in PR #71348:
URL: https://github.com/apache/airflow/pull/71348#discussion_r4451627102
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -3472,36 +3472,55 @@ def adopt_or_reset_orphaned_tasks(self, *, session:
Session = NEW_SESSION) -> in
for executor, tis in exec_to_tis.items():
to_reset.extend(executor.try_adopt_task_instances(tis))
+ # Executors may hand back TaskInstances that are no longer
bound to this
+ # session, and the query above loads only a few columns —
so the reads
+ # below (the full row copied by prepare_db_for_next_try
into
+ # TaskInstanceHistory, last_heartbeat_at, dag_run.conf)
would trigger
+ # lazy loads that raise DetachedInstanceError on a
detached instance and
+ # crash the scheduler loop. Re-select the rows by their
already-loaded
+ # ids so both branches work on fully loaded, session-bound
instances.
+ to_reset_ids = {ti.id for ti in to_reset}
+ ids_to_process = [ti.id for ti in tis_to_adopt_or_reset]
+ bound_tis: list[TaskInstance] = (
+ list(
+ session.scalars(
+
select(TI).options(joinedload(TI.dag_run)).where(TI.id.in_(ids_to_process))
+ )
+ )
+ if ids_to_process
Review Comment:
This always does a second full-row `select(TI)` even when nothing in
`to_reset` actually came back detached. This is probably the right trade-off,
but worth a one-line comment here so a future reader doesn’t reintroduce
load_only on this query without realizing why it was dropped.
Also, `list(session.scalars(...))` should be replaced with
`session.scalars(...).all()` with the annotation changed to
`Sequence[TaskInstance]`.
--
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]
