ashb commented on code in PR #50061:
URL: https://github.com/apache/airflow/pull/50061#discussion_r2124104986
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -1947,12 +1947,13 @@ def _handle_tasks_stuck_in_queued(self, session:
Session = NEW_SESSION) -> None:
for executor, stuck_tis in
self._executor_to_tis(tasks_stuck_in_queued).items():
try:
for ti in stuck_tis:
- executor.revoke_task(ti=ti)
- self._maybe_requeue_stuck_ti(
- ti=ti,
- session=session,
- )
- session.commit()
+ if ti.state == TaskInstanceState.QUEUED:
Review Comment:
I don't think this will actually do anything -- the
`_get_tis_stuck_in_queued` function only returns TIs that are in the queued
state already (via `TI.state == TaskInstanceState.QUEUED`) and we don't go and
refresh the row from the DB in between L1946 and here.
DB transactions (of which we are almost always inside one) will also make
this more tricky. Each DB server has slightly semantics, but on postgresql at
least the default is [Repeatable
Read](https://www.postgresql.org/docs/current/transaction-iso.html#XACT-REPEATABLE-READ):
> The Repeatable Read isolation level only sees data committed before the
transaction began; it never sees either uncommitted data or changes committed
by concurrent transactions during the transaction's execution
We might be able to update `_reschedule_stuck_task` to add `.where(TI.state
== TaskInstanceState.QUEUED)` instead. But that also might be tricky with
transactions, so maybe we need to couple that with `RETURNING` (so that we can
UPDATE anything still in queued, but also get the rows back at the same time)
-- but I don't know if mysql supports that.
In short: erk, this is a lot more complex than it looks like at first.
--
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]
