rjgoyln commented on code in PR #75962:
URL: https://github.com/apache/airflow/pull/75962#discussion_r4417299974
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -3862,23 +3866,49 @@ def check_trigger_timeouts(
"""Mark any "deferred" task as failed if the trigger or execution
timeout has passed."""
for attempt in run_with_db_retries(max_retries, logger=self.log):
with attempt:
- result = session.execute(
- update(TI)
- .where(
- TI.state == TaskInstanceState.DEFERRED,
- TI.trigger_timeout < timezone.utcnow(),
+ try:
+ now = timezone.utcnow()
+ query = (
+ select(TI.id)
+ .where(
+ TI.state == TaskInstanceState.DEFERRED,
+ TI.trigger_timeout < now,
+ )
+ .order_by(TI.id)
+ .limit(MAX_TIMED_OUT_TASK_INSTANCES_PER_LOOP)
)
- .values(
- state=TaskInstanceState.SCHEDULED,
- next_method=TRIGGER_FAIL_REPR,
- next_kwargs={"error":
TriggerFailureReason.TRIGGER_TIMEOUT},
- scheduled_dttm=timezone.utcnow(),
- trigger_id=None,
+ # Claim the rows rather than letting a blind bulk UPDATE
wait on them: a sweep
+ # that waits can be one side of a lock cycle with the
triggerer, which writes
+ # these same rows as tasks defer and fire. SKIP LOCKED
leaves the rows a writer
+ # already holds for the next tick.
+ query = with_row_locks(query, of=TI, session=session,
skip_locked=True)
Review Comment:
Thanks for checking this!
I checked this. supports_for_update_of = False is only the class-level
default; MySQLDialect.initialize() sets it to True on the first connection to a
MySQL 8 server. The guard itself is already covered by test_with_row_locks in
test_sqlalchemy.py.
I also ran a probe against a real MySQL 8.0.46 with both this PR and main.
The PR generates FOR UPDATE OF task_instance SKIP LOCKED, skips locked rows,
and avoids the deadlocks reproduced on main, including in stress tests.
So I'd keep the guard as is. WDYT?
--
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]
