Use SKIP LOCKED instead of NOWAIT in mini scheduler (#39745)

(cherry picked from commit 7dc2b52)
apache · Jun 4, 2024 · e4baada · e4baada
1 parent 7a0fb85
commit e4baada
Showing 1 changed file with 7 additions and 2 deletions.
diff --git a/airflow/models/taskinstance.py b/airflow/models/taskinstance.py
@@ -3455,8 +3455,13 @@ def _schedule_downstream_tasks(
                     run_id=ti.run_id,
                 ),
                 session=session,
-                nowait=True,
-            ).one()
+                skip_locked=True,
+            ).one_or_none()
+
+            if not dag_run:
+                cls.logger().debug("Skip locked rows, rollback")
+                session.rollback()
+                return
 
             task = ti.task
             if TYPE_CHECKING: