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 2 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.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
6 changes: 5 additions & 1 deletion airflow/jobs/scheduler_job_runner.py
View file
Open in desktop
Expand Up @@ -1374,7 +1374,6 @@ 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)
Expand All @@ -1389,6 +1388,11 @@ def _schedule_dag_run(
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 DAG was None due to skip_locked in with_row_locks",
uranusjr marked this conversation as resolved.
Show resolved Hide resolved
dag_run.dag_id,
)
return callback

if (
Expand Down