amoghrajesh commented on code in PR #71072:
URL: https://github.com/apache/airflow/pull/71072#discussion_r3966767217
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -2296,6 +2310,89 @@ def _create_dagruns_for_partitioned_asset_dags(self,
session: Session) -> set[st
if not pending_apdrs:
return set()
Review Comment:
The only place that resets `self._partition_cap_backlog_reported` to False
is line 2338-2340, further down, in the branch that runs when
`len(pending_apdrs) < self._max_partition_dag_runs_per_loop`. That branch never
runs when the backlog is exactly zero, because the function already returned at
line 2311.
So here is the failure case: the backlog builds up, hits the cap, and gets
reported (flag set to True). The backlog then drains all the way to zero in one
tick, the flag stays True forever because the early return at line 2311 skips
the reset. Later, a fresh burst of asset events pushes the backlog straight
back over the cap. The code checks the stale True flag, thinks this is a
continuation of the old episode, logs at debug instead of warning, and skips
writing a new audit Log row entirely (the `if not
self._partition_cap_backlog_reported:` guard at line 2359 is False).
This breaks the one promise this PR makes: "one audit row per backlog
episode." A brand new episode goes completely unaudited if the backlog happened
to touch zero first. Reset the flag at the top of the function whenever
`pending_apdrs` is empty, right next to the early return at line 2310-2311, not
only in the `len(pending_apdrs) < cap` branch further down.
##########
airflow-core/tests/unit/jobs/test_scheduler_job.py:
##########
Review Comment:
No test drains the backlog to literally zero and then rebuilds it. Add a
test that drains the backlog to exactly zero pending APDRs, then adds a fresh
batch that crosses the cap again, and assert the second episode logs at warning
and writes a second audit row.
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -2296,6 +2310,89 @@ def _create_dagruns_for_partitioned_asset_dags(self,
session: Session) -> set[st
if not pending_apdrs:
return set()
+ if len(pending_apdrs) >= self._max_partition_dag_runs_per_loop:
+ # A full fetch alone can't tell us whether that's the entire
backlog or just
+ # this tick's slice of a larger one, so we only pay for this query
then.
+ # Per-Dag counts across the *whole* backlog, not just this tick's
oldest-cap
+ # slice (`pending_apdrs`) — a Dag whose partitions haven't reached
the front of
+ # the FIFO queue yet would otherwise be missing from the log/audit
row until its
+ # turn comes up.
+ backlogs_per_dag: dict[str, int] = {
+ dag_id: count
+ for dag_id, count in session.execute(
+ select(AssetPartitionDagRun.target_dag_id, func.count())
+ .select_from(AssetPartitionDagRun)
+ .join(DagModel, DagModel.dag_id ==
AssetPartitionDagRun.target_dag_id)
+ .where(
+ AssetPartitionDagRun.created_dag_run_id.is_(None),
+ DagModel.is_stale.is_(False),
+ )
+ .group_by(AssetPartitionDagRun.target_dag_id)
+ )
+ }
+ backlog_total = sum(backlogs_per_dag.values())
+ # No SQL-side ORDER BY: it would have no dag_id tie-break, and
relying on dict
+ # insertion order mirroring DB row order across backends (SQLite
in tests vs
+ # Postgres/MySQL in production) isn't a contract worth trusting —
sort explicitly.
+ sorted_dag_ids = sorted(backlogs_per_dag, key=lambda dag_id:
(-backlogs_per_dag[dag_id], dag_id))
+ else:
+ backlog_total = len(pending_apdrs)
+ self._partition_cap_backlog_reported = False
+
+ if backlog_total > self._max_partition_dag_runs_per_loop:
+ displayed_dag_ids =
sorted_dag_ids[:MAX_PARTITION_CAP_BACKLOG_DAG_IDS_LOGGED]
+ remaining_dag_id_count = len(sorted_dag_ids) -
len(displayed_dag_ids)
+ log_cap_reached = self.log.debug if
self._partition_cap_backlog_reported else self.log.warning
Review Comment:
`sorted_dag_ids` is set in a conditional statement above and it works fine
today but it depends on two separate if conditions staying in sync by
coincidence, not by a name or a comment saying so. A future edit to either
condition (e.g. changing `>=` to `>`, or changing what backlog_total means in
the else branch) can turn this into a `NameError`
--
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]
