ataulmujeeb-cyber opened a new pull request, #76991:
URL: https://github.com/apache/airflow/pull/76991
Partitioned asset consumers can wait indefinitely without creating a Dag
run. After an upstream dependency is added, an event published before that
dependency existed is not attached to the pending consumer. Alerts tied to an
existing run cannot surface that wait.
Add an opt-in `scheduler.asset_partition_pending_timeouts` setting, mapping
exact Dag IDs to their maximum expected pending age in seconds:
```ini
[scheduler]
asset_partition_pending_timeouts = {"daily_sales": 3600}
```
The scheduler records an `asset partition pending` warning in the existing
Dag warning UI and `/api/v2/dagWarnings` API when the oldest pending row
exceeds that threshold. The message includes the pending row count, oldest
creation time and configured threshold. It clears when the pending work is
resolved, the Dag becomes ineligible, or its policy is removed.
Checks run every minute, rotating through at most 100 Dag IDs per
transaction. Paused, draining and stale Dags are excluded; paused time still
counts toward age after resuming. Multiple schedulers use atomic upserts,
preserve the first-detected timestamp and leave unchanged warnings alone. All
schedulers should use the same policy. Warning updates are eventually
consistent.
This monitors existing pending rows. It does not detect a missed cadence
with no pending row, send notifications, or replay events. Scheduling and the
database schema are unchanged.
Related to the [monitoring discussion in
#76979](https://github.com/apache/airflow/issues/76979#issuecomment-6833027935)
and #30974. This is separate from recovery in #76980, pending-work rotation in
#76463, resynchronization in #76706, and cap-pressure reporting in #71072.
### Validation
- PostgreSQL 16 / psycopg 3: 94 tests passed for
`test_asset_partition_warnings.py` and the partition selection in
`test_scheduler_job.py` (`-k partition`). A separate 64-test selection passed
for the monitor, timer registration, dependency-change scenario, parser warning
retention, public warning API, and execution API version boundary.
- MySQL 8.4 / PyMySQL: 36 passed for the monitor and dependency-change
integration.
- SQLite: the same selection passed 35 tests, with the concurrent-monitor
test skipped.
- The dependency-change test publishes B before the consumer depends on B,
changes A to A&B, then publishes A. It verifies no consumer run is created, the
warning appears, and a fresh B event lets the real scheduler create the run and
clear the warning. Replacing warning refresh with a no-op makes this test fail
at the missing-warning assertion.
- Coverage includes the exact threshold boundary, repeated partition keys,
more than 500 pending rows, bounded rotation, rollback, concurrent schedulers,
configuration removal and unrelated warnings.
- Ruff, formatting, full distribution mypy checks for core/Task
SDK/airflowctl, manual hooks, default configuration validation, application
TypeScript checks and the UI build passed. API schemas and clients were
regenerated. Two local hook wrappers were excluded after reproducing their
failures: the config wrapper invokes a broken global Breeze launcher; the UI
wrapper asks oxlint to lint ignored generated files. Default configuration
validation ran directly. API generation, application TypeScript checks and the
UI build passed separately; the UI lint hook remained blocked because oxlint
excludes the generated files.
These checks ran on `0fa3336929`. The published commit `8d6b819865` is a
conflict-free rebase over a provider-metadata-only main update; `git
range-diff` confirms the patch is unchanged.
The principal selections can be reproduced with the repository test
environment:
```sh
uv run --project airflow-core pytest
airflow-core/tests/unit/jobs/test_asset_partition_warnings.py
airflow-core/tests/unit/jobs/test_scheduler_job.py -k partition
uv run --project airflow-core pytest
airflow-core/tests/unit/jobs/test_asset_partition_warnings.py
airflow-core/tests/unit/jobs/test_scheduler_job.py::test_partition_wait_warning_after_dependency_edit_clears_when_real_run_is_created
uv run --project airflow-core pytest
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_warning.py
```
### Polling cost
A local PostgreSQL comparison used three alternating base/head pairs,
100,000 completed rows and 1,590 pending rows across 100 Dags. Median steady
callback times were 1.4 ms with monitoring disabled, 2.9 ms for one monitored
Dag and 8.6 ms for 100 Dags. The base/head control medians differed by at most
9.3% across cases. The base has no callback, so these are added costs, not
slowdown ratios. These measurements are not a production capacity claim; the
aggregation can inspect historical rows for the selected Dags.
The broader cross-package suites and documentation build are left to CI; the
results above describe the local selections. The local docs build requires
Breeze's Python 3.11 image, which was not installed.
--
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]
