You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "seanmuth (via GitHub)" <gi...@apache.org> on 2026/09/24 19:54:46 UTC
[PR] Log when a newly created Dag Run will not be scheduled due to max_active_runs [airflow]
seanmuth opened a new pull request, #73689:
URL: https://github.com/apache/airflow/pull/73689
When a Dag Run is created while its Dag is already at `max_active_runs` (for example, a second manual trigger while an earlier run is still active), no log message anywhere explains why it never starts. The scheduler's periodic max_active_runs check (enhanced by #72403) only logs the moment a Dag *newly* becomes blocked — not for a run created afterward while it's already blocked, since the promotion query filters it out before that code path is ever reached.
This adds a one-shot info log at `DagRun` creation time (`SerializedDAG.create_dagrun`) when the new run will not be immediately schedulable, so the information is available right away without correlating scheduler internals.
Part of #73686, which tracks the two related follow-ups (exposing this state via the API, and a UI tooltip) as separate PRs.
related: #72387, #72403
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes — Claude Sonnet 5
Generated-by: Claude Sonnet 5 following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Log when a newly created Dag Run will not be scheduled due to max_active_runs [airflow]
Posted by "seanmuth (via GitHub)" <gi...@apache.org>.
seanmuth commented on code in PR #73689:
URL: https://github.com/apache/airflow/pull/73689#discussion_r4124440437
##########
airflow-core/src/airflow/models/dagrun.py:
##########
@@ -744,6 +744,47 @@ def active_runs_of_dags(
query = query.where(cls.run_type != DagRunType.BACKFILL_JOB)
return {dag_id: count for dag_id, count in session.execute(query)}
+ @classmethod
+ @provide_session
+ def log_if_new_run_blocked_by_max_active_runs(
+ cls,
+ *,
+ dag: SerializedDAG,
+ run_id: str,
+ session: Session = NEW_SESSION,
+ ) -> None:
+ """
+ Log if a just-created DagRun will not be scheduled yet because the Dag is at max_active_runs.
+
+ Meant to be called by request-driven trigger surfaces -- manual UI/REST-API triggers and
+ TriggerDagRunOperator/CLI (via :func:`airflow.api.common.trigger_dag.trigger_dag`) -- right
+ after :meth:`SerializedDAG.create_dagrun`. Deliberately not called from the scheduler's own
+ run-creation call sites: those run every scheduling loop, and the scheduler already has
+ separate periodic bookkeeping for this (``_set_exceeds_max_active_runs``) that intentionally
Review Comment:
You're right — confirmed `_set_exceeds_max_active_runs` doesn't log anything on `main`. I'd conflated it with #72403, which was closed without merging. Reworded the docstring to drop that reference; the reasoning for keeping this out of the scheduler (its own creation paths run every loop) stands fine without it. Also caught the same stale #72403-as-merged framing in the tracking issue's body and corrected it there too.
---
Drafted-by: Claude Sonnet 5; reviewed by @seanmuth before posting
##########
airflow-core/tests/unit/models/test_dagrun.py:
##########
@@ -1649,6 +1649,86 @@ def test_dagrun_deadline_variable_interval_missing_variable_fails(self, _, sessi
session=session,
)
+ def test_log_if_new_run_blocked_by_max_active_runs_logs_when_at_max(self, session, caplog):
+ dag = DAG(
+ dag_id="test_log_if_new_run_blocked_by_max_active_runs_logs_when_at_max",
+ schedule=None,
+ max_active_runs=1,
+ )
+ scheduler_dag = sync_dag_to_db(dag, session=session)
+ scheduler_dag.create_dagrun(
+ run_id="running_run",
+ logical_date=DEFAULT_DATE,
+ data_interval=(DEFAULT_DATE, DEFAULT_DATE),
+ run_after=DEFAULT_DATE,
+ run_type=DagRunType.MANUAL,
+ state=DagRunState.RUNNING,
+ triggered_by=DagRunTriggeredByType.TEST,
+ session=session,
+ )
+
+ with caplog.at_level("INFO", logger="airflow.models.dagrun"):
+ DagRun.log_if_new_run_blocked_by_max_active_runs(
+ dag=scheduler_dag, run_id="queued_run", session=session
+ )
+
+ assert {
+ "event": "created DagRun will not be scheduled yet, dag is at max_active_runs",
+ "dag_id": "test_log_if_new_run_blocked_by_max_active_runs_logs_when_at_max",
+ "run_id": "queued_run",
+ "active_runs": 1,
+ "max_active_runs": 1,
+ "log_level": "info",
+ } in caplog
+
+ def test_log_if_new_run_blocked_by_max_active_runs_does_not_log_when_below_max(self, session, caplog):
Review Comment:
Done — consolidated into one parametrized test (also added a 4th case covering the backfill fix above) and switched the negative assertions to the `not in caplog` membership form instead of `caplog.records`.
---
Drafted-by: Claude Sonnet 5; reviewed by @seanmuth before posting
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Log when a newly created Dag Run will not be scheduled due to max_active_runs [airflow]
Posted by "seanmuth (via GitHub)" <gi...@apache.org>.
seanmuth commented on code in PR #73689:
URL: https://github.com/apache/airflow/pull/73689#discussion_r4134738071
##########
airflow-core/src/airflow/models/dagrun.py:
##########
@@ -744,6 +744,54 @@ def active_runs_of_dags(
query = query.where(cls.run_type != DagRunType.BACKFILL_JOB)
return {dag_id: count for dag_id, count in session.execute(query)}
+ @classmethod
+ @provide_session
+ def log_if_new_run_blocked_by_max_active_runs(
+ cls,
+ *,
+ dag: SerializedDAG,
+ run_id: str,
+ session: Session = NEW_SESSION,
+ ) -> None:
+ """
+ Log if a just-created DagRun will not be scheduled yet because the Dag is at max_active_runs.
+
+ Meant to be called by request-driven trigger surfaces -- manual UI/REST-API triggers and
+ TriggerDagRunOperator/CLI (via :func:`airflow.api.common.trigger_dag.trigger_dag`) -- right
+ after :meth:`SerializedDAG.create_dagrun`. Deliberately not called from the scheduler's own
+ run-creation call sites: those run every scheduling loop, so calling this from there too
+ would turn a one-shot, per-trigger log into a per-loop one instead.
+
+ Counts running non-backfill runs only, to match the promotion queries
+ (:meth:`get_queued_dag_runs_to_set_running` and the scheduler's ``_start_queued_dagruns``),
+ which both count running runs per ``(dag_id, backfill_id)`` -- a running backfill run
+ doesn't hold back a manual trigger's own max_active_runs slot.
+
+ :meta private:
+ """
+ if not dag.max_active_runs:
Review Comment:
Good catch on both counts, fixed. Converted this to an instance method on the just-created `DagRun` — it now reads `self.max_active_runs` (the `dag_model`-backed association proxy, same source `_start_queued_dagruns` uses via `dag_run.max_active_runs`) instead of the serialized Dag's value, and drops the `dag`/`run_id` params entirely.
Also dropped the early `if not dag.max_active_runs: return` and let the comparison run unconditionally as `num_running >= (self.max_active_runs or 0)`, so `max_active_runs=0` now logs correctly instead of being treated as unset. Updated the `max_active_runs_unset` test case to `max_active_runs_zero_never_promoted` with `expect_log=True` to lock that in, and the other three call sites (`trigger_dag.py`, and the two API routes) now call `dag_run.log_if_new_run_blocked_by_max_active_runs()` on the run they just created.
---
Drafted-by: Claude Sonnet 5 (no human review before posting)
##########
airflow-core/tests/unit/models/test_dagrun.py:
##########
@@ -1649,6 +1649,65 @@ def test_dagrun_deadline_variable_interval_missing_variable_fails(self, _, sessi
session=session,
)
+ @pytest.mark.parametrize(
+ ("max_active_runs", "running_run_type", "expect_log"),
+ [
+ pytest.param(1, "manual", True, id="at_max"),
+ pytest.param(1, None, False, id="below_max"),
+ pytest.param(0, "manual", False, id="max_active_runs_unset"),
+ pytest.param(1, "backfill", False, id="running_backfill_run_does_not_count"),
+ ],
+ )
+ def test_log_if_new_run_blocked_by_max_active_runs(
+ self, session, caplog, max_active_runs, running_run_type, expect_log
+ ):
+ from airflow.models.backfill import Backfill
Review Comment:
Moved to the top-level imports, thanks.
---
Drafted-by: Claude Sonnet 5 (no human review before posting)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Log when a newly created Dag Run will not be scheduled due to max_active_runs [airflow]
Posted by "seanmuth (via GitHub)" <gi...@apache.org>.
seanmuth commented on code in PR #73689:
URL: https://github.com/apache/airflow/pull/73689#discussion_r4108293049
##########
airflow-core/src/airflow/serialization/definitions/dag.py:
##########
@@ -712,6 +712,24 @@ def create_dagrun(
if params_dag.deadline:
self._process_dagrun_deadline_alerts(orm_dagrun, session)
+ if state == DagRunState.QUEUED and backfill_id is None and self.max_active_runs:
+ num_running = (
+ session.scalar(
+ select(func.count())
+ .select_from(DagRun)
+ .where(DagRun.dag_id == self.dag_id, DagRun.state == DagRunState.RUNNING)
+ )
+ or 0
+ )
+ if num_running >= self.max_active_runs:
+ log.info(
+ "created DagRun will not be scheduled yet, dag is at max_active_runs",
+ dag_id=self.dag_id,
+ run_id=run_id,
+ active_runs=num_running,
+ max_active_runs=self.max_active_runs,
+ )
+
Review Comment:
agreed. made a helpful function and moved it's invocation to the 3 call-sites outside of the scheduler loop.
##########
airflow-core/src/airflow/serialization/definitions/dag.py:
##########
@@ -712,6 +712,24 @@ def create_dagrun(
if params_dag.deadline:
self._process_dagrun_deadline_alerts(orm_dagrun, session)
+ if state == DagRunState.QUEUED and backfill_id is None and self.max_active_runs:
+ num_running = (
+ session.scalar(
+ select(func.count())
+ .select_from(DagRun)
+ .where(DagRun.dag_id == self.dag_id, DagRun.state == DagRunState.RUNNING)
+ )
+ or 0
+ )
+ if num_running >= self.max_active_runs:
+ log.info(
+ "created DagRun will not be scheduled yet, dag is at max_active_runs",
+ dag_id=self.dag_id,
+ run_id=run_id,
+ active_runs=num_running,
+ max_active_runs=self.max_active_runs,
+ )
+
Review Comment:
agreed. made a helpfer function and moved it's invocation to the 3 call-sites outside of the scheduler loop.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Log when a newly created Dag Run will not be scheduled due to max_active_runs [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #73689:
URL: https://github.com/apache/airflow/pull/73689#discussion_r4109598256
##########
airflow-core/src/airflow/models/dagrun.py:
##########
@@ -744,6 +744,47 @@ def active_runs_of_dags(
query = query.where(cls.run_type != DagRunType.BACKFILL_JOB)
return {dag_id: count for dag_id, count in session.execute(query)}
+ @classmethod
+ @provide_session
+ def log_if_new_run_blocked_by_max_active_runs(
+ cls,
+ *,
+ dag: SerializedDAG,
+ run_id: str,
+ session: Session = NEW_SESSION,
+ ) -> None:
+ """
+ Log if a just-created DagRun will not be scheduled yet because the Dag is at max_active_runs.
+
+ Meant to be called by request-driven trigger surfaces -- manual UI/REST-API triggers and
+ TriggerDagRunOperator/CLI (via :func:`airflow.api.common.trigger_dag.trigger_dag`) -- right
+ after :meth:`SerializedDAG.create_dagrun`. Deliberately not called from the scheduler's own
+ run-creation call sites: those run every scheduling loop, and the scheduler already has
+ separate periodic bookkeeping for this (``_set_exceeds_max_active_runs``) that intentionally
+ only logs once when a Dag *newly* becomes blocked, rather than every loop. Calling this from
+ there too would turn a one-shot, per-trigger log back into that same per-loop spam.
+
+ :meta private:
+ """
+ if not dag.max_active_runs:
+ return
+ num_running = (
+ session.scalar(
+ select(func.count())
+ .select_from(cls)
+ .where(cls.dag_id == dag.dag_id, cls.state == DagRunState.RUNNING)
Review Comment:
This counts every running run of the Dag, backfill runs included, but the scheduler doesn't. Both `get_queued_dag_runs_to_set_running` and `_start_queued_dagruns` count running runs per `(dag_id, backfill_id)`, so a manual run is only held back by other non-backfill runs. With a backfill in flight on a `max_active_runs=1` Dag, a manual trigger would log "will not be scheduled yet" and then start on the next loop, which is the misleading signal this PR is trying to remove. Adding `cls.backfill_id.is_(None)` here, plus a test with a running backfill run, would line it up with the promotion query.
##########
airflow-core/tests/unit/models/test_dagrun.py:
##########
@@ -1649,6 +1649,86 @@ def test_dagrun_deadline_variable_interval_missing_variable_fails(self, _, sessi
session=session,
)
+ def test_log_if_new_run_blocked_by_max_active_runs_logs_when_at_max(self, session, caplog):
+ dag = DAG(
+ dag_id="test_log_if_new_run_blocked_by_max_active_runs_logs_when_at_max",
+ schedule=None,
+ max_active_runs=1,
+ )
+ scheduler_dag = sync_dag_to_db(dag, session=session)
+ scheduler_dag.create_dagrun(
+ run_id="running_run",
+ logical_date=DEFAULT_DATE,
+ data_interval=(DEFAULT_DATE, DEFAULT_DATE),
+ run_after=DEFAULT_DATE,
+ run_type=DagRunType.MANUAL,
+ state=DagRunState.RUNNING,
+ triggered_by=DagRunTriggeredByType.TEST,
+ session=session,
+ )
+
+ with caplog.at_level("INFO", logger="airflow.models.dagrun"):
+ DagRun.log_if_new_run_blocked_by_max_active_runs(
+ dag=scheduler_dag, run_id="queued_run", session=session
+ )
+
+ assert {
+ "event": "created DagRun will not be scheduled yet, dag is at max_active_runs",
+ "dag_id": "test_log_if_new_run_blocked_by_max_active_runs_logs_when_at_max",
+ "run_id": "queued_run",
+ "active_runs": 1,
+ "max_active_runs": 1,
+ "log_level": "info",
+ } in caplog
+
+ def test_log_if_new_run_blocked_by_max_active_runs_does_not_log_when_below_max(self, session, caplog):
Review Comment:
These three tests share the same setup and differ only in `max_active_runs`, whether a running run exists, and the expected outcome, so one `pytest.mark.parametrize` test would cover them. The negative checks can also be `"created DagRun will not be scheduled yet, dag is at max_active_runs" not in caplog`, since `caplog.records` on the structlog capture is marked for removal in `tests_common`.
##########
airflow-core/src/airflow/models/dagrun.py:
##########
@@ -744,6 +744,47 @@ def active_runs_of_dags(
query = query.where(cls.run_type != DagRunType.BACKFILL_JOB)
return {dag_id: count for dag_id, count in session.execute(query)}
+ @classmethod
+ @provide_session
+ def log_if_new_run_blocked_by_max_active_runs(
+ cls,
+ *,
+ dag: SerializedDAG,
+ run_id: str,
+ session: Session = NEW_SESSION,
+ ) -> None:
+ """
+ Log if a just-created DagRun will not be scheduled yet because the Dag is at max_active_runs.
+
+ Meant to be called by request-driven trigger surfaces -- manual UI/REST-API triggers and
+ TriggerDagRunOperator/CLI (via :func:`airflow.api.common.trigger_dag.trigger_dag`) -- right
+ after :meth:`SerializedDAG.create_dagrun`. Deliberately not called from the scheduler's own
+ run-creation call sites: those run every scheduling loop, and the scheduler already has
+ separate periodic bookkeeping for this (``_set_exceeds_max_active_runs``) that intentionally
Review Comment:
`_set_exceeds_max_active_runs` doesn't log anything on main. The "only logs once when a Dag newly becomes blocked" behaviour came from #72403, which was closed without merging. The reason for keeping this out of the scheduler (its creation paths run every loop) holds on its own, so I'd drop the reference to that bookkeeping or reword it to describe what's actually there.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Log when a newly created Dag Run will not be scheduled due to max_active_runs [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #73689:
URL: https://github.com/apache/airflow/pull/73689#discussion_r4127780586
##########
airflow-core/tests/unit/models/test_dagrun.py:
##########
@@ -1649,6 +1649,65 @@ def test_dagrun_deadline_variable_interval_missing_variable_fails(self, _, sessi
session=session,
)
+ @pytest.mark.parametrize(
+ ("max_active_runs", "running_run_type", "expect_log"),
+ [
+ pytest.param(1, "manual", True, id="at_max"),
+ pytest.param(1, None, False, id="below_max"),
+ pytest.param(0, "manual", False, id="max_active_runs_unset"),
+ pytest.param(1, "backfill", False, id="running_backfill_run_does_not_count"),
+ ],
+ )
+ def test_log_if_new_run_blocked_by_max_active_runs(
+ self, session, caplog, max_active_runs, running_run_type, expect_log
+ ):
+ from airflow.models.backfill import Backfill
Review Comment:
No circular import here (other test modules import `Backfill` at module level), so this can move up to the top-level `airflow.models` imports.
##########
airflow-core/src/airflow/models/dagrun.py:
##########
@@ -744,6 +744,54 @@ def active_runs_of_dags(
query = query.where(cls.run_type != DagRunType.BACKFILL_JOB)
return {dag_id: count for dag_id, count in session.execute(query)}
+ @classmethod
+ @provide_session
+ def log_if_new_run_blocked_by_max_active_runs(
+ cls,
+ *,
+ dag: SerializedDAG,
+ run_id: str,
+ session: Session = NEW_SESSION,
+ ) -> None:
+ """
+ Log if a just-created DagRun will not be scheduled yet because the Dag is at max_active_runs.
+
+ Meant to be called by request-driven trigger surfaces -- manual UI/REST-API triggers and
+ TriggerDagRunOperator/CLI (via :func:`airflow.api.common.trigger_dag.trigger_dag`) -- right
+ after :meth:`SerializedDAG.create_dagrun`. Deliberately not called from the scheduler's own
+ run-creation call sites: those run every scheduling loop, so calling this from there too
+ would turn a one-shot, per-trigger log into a per-loop one instead.
+
+ Counts running non-backfill runs only, to match the promotion queries
+ (:meth:`get_queued_dag_runs_to_set_running` and the scheduler's ``_start_queued_dagruns``),
+ which both count running runs per ``(dag_id, backfill_id)`` -- a running backfill run
+ doesn't hold back a manual trigger's own max_active_runs slot.
+
+ :meta private:
+ """
+ if not dag.max_active_runs:
Review Comment:
The backfill fix lines up the count, but the limit still comes from a different place than the gate uses. Both promotion paths read `DagModel.max_active_runs` (`_start_queued_dagruns` goes through `dag_run.max_active_runs`, with a comment about not trusting a stale serialized Dag), and the serialized write can lag `DagModel` by up to `min_serialized_dag_update_interval` after someone edits the value. `max_active_runs=0` is also the one case where the run is never promoted (the prefilter is `0 < 0`), and this early return skips logging it; the `max_active_runs_unset` test case locks that in. Making this an instance method that reads `self.max_active_runs or 0` off the just-created run would fix both and drop the `dag`/`run_id` params.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Log when a newly created Dag Run will not be scheduled due to max_active_runs [airflow]
Posted by "seanmuth (via GitHub)" <gi...@apache.org>.
seanmuth commented on code in PR #73689:
URL: https://github.com/apache/airflow/pull/73689#discussion_r4124438601
##########
airflow-core/src/airflow/models/dagrun.py:
##########
@@ -744,6 +744,47 @@ def active_runs_of_dags(
query = query.where(cls.run_type != DagRunType.BACKFILL_JOB)
return {dag_id: count for dag_id, count in session.execute(query)}
+ @classmethod
+ @provide_session
+ def log_if_new_run_blocked_by_max_active_runs(
+ cls,
+ *,
+ dag: SerializedDAG,
+ run_id: str,
+ session: Session = NEW_SESSION,
+ ) -> None:
+ """
+ Log if a just-created DagRun will not be scheduled yet because the Dag is at max_active_runs.
+
+ Meant to be called by request-driven trigger surfaces -- manual UI/REST-API triggers and
+ TriggerDagRunOperator/CLI (via :func:`airflow.api.common.trigger_dag.trigger_dag`) -- right
+ after :meth:`SerializedDAG.create_dagrun`. Deliberately not called from the scheduler's own
+ run-creation call sites: those run every scheduling loop, and the scheduler already has
+ separate periodic bookkeeping for this (``_set_exceeds_max_active_runs``) that intentionally
+ only logs once when a Dag *newly* becomes blocked, rather than every loop. Calling this from
+ there too would turn a one-shot, per-trigger log back into that same per-loop spam.
+
+ :meta private:
+ """
+ if not dag.max_active_runs:
+ return
+ num_running = (
+ session.scalar(
+ select(func.count())
+ .select_from(cls)
+ .where(cls.dag_id == dag.dag_id, cls.state == DagRunState.RUNNING)
Review Comment:
Good catch — you're right, this counted every running run regardless of `backfill_id`, which doesn't match how the promotion queries actually gate concurrency. Added `cls.backfill_id.is_(None)` and a test with a running backfill run present, to lock in the fix — pushed.
---
Drafted-by: Claude Sonnet 5; reviewed by @seanmuth before posting
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Log when a newly created Dag Run will not be scheduled due to max_active_runs [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #73689:
URL: https://github.com/apache/airflow/pull/73689#discussion_r4103245440
##########
airflow-core/src/airflow/serialization/definitions/dag.py:
##########
@@ -712,6 +712,24 @@ def create_dagrun(
if params_dag.deadline:
self._process_dagrun_deadline_alerts(orm_dagrun, session)
+ if state == DagRunState.QUEUED and backfill_id is None and self.max_active_runs:
+ num_running = (
+ session.scalar(
+ select(func.count())
+ .select_from(DagRun)
+ .where(DagRun.dag_id == self.dag_id, DagRun.state == DagRunState.RUNNING)
+ )
+ or 0
+ )
+ if num_running >= self.max_active_runs:
+ log.info(
+ "created DagRun will not be scheduled yet, dag is at max_active_runs",
+ dag_id=self.dag_id,
+ run_id=run_id,
+ active_runs=num_running,
+ max_active_runs=self.max_active_runs,
+ )
+
Review Comment:
I don't know about this living in here. My gut says this belongs with the caller, not here. (create_dagrun did what it was told, it created it. That it won't be scheduled is nothing to do with creating a dagrun)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org