Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Fix future DagRun rarely triggered by race conditions when max_active_runs reached its upper limit. #31414

Merged
merged 8 commits into from
Aug 8, 2023
Merged

Fix future DagRun rarely triggered by race conditions when max_active_runs reached its upper limit. #31414

Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
035826e
feat: select dag_model with row lock
doiken May 19, 2023
82c976a
fix: logging that scheduling was skipped
doiken May 23, 2023
8492e3d
fix: remove unused get_dagmodel
doiken May 23, 2023
fe6df83
fix: correct log message to more generic word
doiken May 24, 2023
d807dff
Merge branch 'main' into fix-triggering-future-dagrun
uranusjr May 25, 2023
1d78cdc
Merge branch 'main' into fix-triggering-future-dagrun
uranusjr Jun 14, 2023
625ebba
Merge branch 'main' into fix-triggering-future-dagrun
doiken Jul 12, 2023
9785780
Merge branch 'main' into fix-triggering-future-dagrun
eladkal Aug 8, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 16 additions & 3 deletions airflow/jobs/scheduler_job_runner.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,7 @@
from sqlalchemy import and_, delete, func, not_, or_, select, text, update
from sqlalchemy.engine import Result
from sqlalchemy.exc import OperationalError
from sqlalchemy.orm import Query, Session, load_only, make_transient, selectinload
from sqlalchemy.orm import Query, Session, joinedload, load_only, make_transient, selectinload
from sqlalchemy.sql import expression

from airflow import settings
Expand Down Expand Up @@ -1397,11 +1397,24 @@ def _schedule_dag_run(
callback: DagCallbackRequest | None = None

dag = dag_run.dag = self.dagbag.get_dag(dag_run.dag_id, session=session)
dag_model = DM.get_dagmodel(dag_run.dag_id, session)
# Adopt row locking to account for inconsistencies when next_dagrun_create_after = None
query = (
session.query(DagModel)
.filter(DagModel.dag_id == dag_run.dag_id)
.options(joinedload(DagModel.parent_dag))
)
dag_model = with_row_locks(
query, of=DagModel, session=session, **skip_locked(session=session)
).one_or_none()

if not dag or not dag_model:
if not dag:
self.log.error("Couldn't find DAG %s in DAG bag or database!", dag_run.dag_id)
return callback
if not dag_model:
self.log.info(
"DAG %s scheduling was skipped, probably because the DAG record was locked", dag_run.dag_id
)
return callback

if (
dag_run.start_date
Expand Down