ataulmujeeb-cyber opened a new pull request, #76980:
URL: https://github.com/apache/airflow/pull/76980
Adding an asset to a partitioned consumer can leave the current partition
missing an input that was published before the dependency existed. Republishing
that input currently queues work for every eligible consumer, including
consumers that already processed the partition.
This adds an optional `consumer_dag_id` to `POST /api/v2/assets/events`.
Operators can publish the missing input for one eligible consumer without
queueing another run for its siblings. The caller must have both asset-event
creation permission and run creation permission for the selected Dag. Normal
consumer resolution, team restrictions, paused/draining state and partition
mapping still apply.
The request creates a new event. Asset history, listeners and metrics retain
their normal behavior; historical events are not attached retroactively. If
pending progress belongs to an old dependency or mapper definition, the
documented recovery is to clear the affected consumer's queued inputs and
publish scoped events for every required input under the new definition. Queue
deletion filters source partition keys, so non-identity mappings require care.
This does not change automatic historical catchup or introduce a migration.
Closes #76979. Related to #76706: reevaluating an existing pending run
cannot supply an input link that was never created.
Validation against main at `29331a8325698266e380a88e6aece1dad7669210`:
- PostgreSQL 16, psycopg 3.3.8, Python 3.13: **347 passed** across
`airflow-core/tests/unit/assets/test_manager.py` and
`airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py`.
- The two recovery tests call the public HTTP routes, serialize the changed
consumer definition, and run the scheduler. They check exact consumed event
IDs, unchanged original event fields, and no additional sibling run or pending
partition. Removing only the consumer filter makes both tests fail because the
sibling is triggered.
- Manager/API coverage includes direct, alias, name and URI consumer
resolution; omitted/null scope; target authorization; team filtering; invalid,
paused and draining targets; and partition-key compatibility.
- Core and AirflowCTL type checks, Ruff, regular static checks and manual
checks were run. The OpenAPI schema and Python/UI models were regenerated;
OpenAPI validation, the UI TypeScript check and the production UI build passed.
The focused runtime command is:
```bash
uv run --project airflow-core pytest \
airflow-core/tests/unit/assets/test_manager.py \
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py
-q
```
The database run used an isolated PostgreSQL database, the workspace sources
and the matching PostgreSQL driver. The broader Breeze matrix was not run
locally because the CI-image build exhausted Docker's available disk space. The
OpenAPI generator ran directly under the supported workspace environment
because the installed Breeze launcher was broken. The UI hook's generated-file
lint step reports "No files found to lint" because those paths are deliberately
ignored by its linter; code generation and the production build were run
separately. These are local validation limits, not CI passes.
--
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]
