ataulmujeeb-cyber opened a new issue, #76979:
URL: https://github.com/apache/airflow/issues/76979
Changing a partitioned consumer from `A` to `A & B` can leave it without
usable input for a partition even when both assets have published that
partition. Republishing B currently queues work for other B consumers too. An
explicit recovery operation should allow queueing an event only for the
affected consumer and partition.
There are two distinct cases to account for. The source behavior below is
from main at `29331a8325698266e380a88e6aece1dad7669210`; runtime verification
is recorded separately below.
### Reproduction sketch
```python
from airflow.sdk import DAG, Asset, IdentityMapper, PartitionedAssetTimetable
from airflow.providers.standard.operators.empty import EmptyOperator
a = Asset(name="reconcile-a", uri="test://reconcile/a")
b = Asset(name="reconcile-b", uri="test://reconcile/b")
with DAG(
dag_id="target",
schedule=PartitionedAssetTimetable(
assets=a, # Later change this to a & b.
default_partition_mapper=IdentityMapper(),
batch_asset_events=True,
),
catchup=False,
):
EmptyOperator(task_id="consume")
with DAG(
dag_id="other-b-consumer",
schedule=PartitionedAssetTimetable(
assets=b,
default_partition_mapper=IdentityMapper(),
),
catchup=False,
):
EmptyOperator(task_id="consume")
```
Deploy and unpause both consumers, and wait for the definitions to be
serialized and the assets to be active. Publish events through `POST
/api/v2/assets/events` using the registered asset IDs and bodies such as
`{"asset_id": <B_ID>, "partition_key": "k"}`.
**Case 1: B published before the dependency existed.**
1. Publish B/k and allow `other-b-consumer` to consume it.
2. Change `target` from A to A & B, then wait for the new definition to be
serialized.
3. Publish A/k and let the scheduler evaluate it repeatedly.
4. Inspect the pending partition run and its input links. B's historical
event has no link to this consumer's pending run, so reevaluation alone cannot
supply B.
5. Republishing B/k supplies a new event to `target`, but also queues work
for `other-b-consumer` again.
**Case 2: a pending partition run predates the dependency change.**
Starting from a clean state with `target` depending only on A, publish A/k
and retain the pending partition run before the scheduler claims it. Change the
definition to A & B and serialize it, then run the scheduler. Adding B changes
the rollup fingerprint, so stale-fingerprint cleanup removes the old pending
run and its input links. This case is discarded progress, not a pending run
that simply waits forever. The underlying global asset events remain.
### Source findings and recovery requirements
The scheduler evaluates inputs from [`PartitionedAssetKeyLog` rows belonging
to each pending
run](https://github.com/apache/airflow/blob/29331a8325698266e380a88e6aece1dad7669210/airflow-core/src/airflow/jobs/scheduler_job_runner.py#L2640),
rather than searching all historical asset events. The [fingerprint includes
each referenced asset and
mapper](https://github.com/apache/airflow/blob/29331a8325698266e380a88e6aece1dad7669210/airflow-core/src/airflow/timetables/base.py#L553),
so a dependency addition reaches the [stale-run cleanup
path](https://github.com/apache/airflow/blob/29331a8325698266e380a88e6aece1dad7669210/airflow-core/src/airflow/jobs/scheduler_job_runner.py#L2566).
The proposed recovery is an optional `consumer_dag_id` on `POST
/api/v2/assets/events`. It would create a new asset event and queue it only for
the selected eligible consumer. The event would retain the usual global
history, listeners, and metrics; it would not reuse a historical event's
identity or bypass access controls, asset-team restrictions, paused/draining
state, or partition mapping. The example uses `catchup=False` and does not
assume automatic historical replay is already promised.
For case 1, publishing B/k with the selected consumer would supply the
missing input. For case 2, first clear the affected consumer's old queued
inputs, then publish scoped events for all inputs needed under the new
definition. Queue deletion matches source partition keys, which can differ from
target keys under a mapper. Partial clearing retains the remaining pending
progress, so recovery must account for every contributing source key.
A blanket history scan would be unsafe. Queued partition inputs can now be
intentionally deleted through the API (#73119); silently restoring their links
would undo that operation. Previously consumed events, changed partition
mappings, and inactive assets also need defined treatment.
Related work: #76706 wakes pending partition runs when their inputs or
definitions change. Reevaluation of existing links does not supply a missing
historical link, so that work alone does not cover this recovery case.
### Verification
Reproduced against unmodified production code at the main commit above with
PostgreSQL 16, psycopg 3.3.8 and Python 3.13. The controls use actual Dag
serialization/dependency updates, API-style event registration through
`AssetManager`, and real scheduler ticks.
- In case 1, repeated ticks leave the target pending. Republishing B creates
runs for both target and sibling; their consumed event IDs identify the
replacement B event.
- In case 2, a fresh B event attaches to the old candidate before the tick.
The tick removes that candidate and both input links, creates no target run,
and retains both global events.
I am preparing a PR for consumer-scoped event publication. It will provide
explicit recovery rather than changing automatic historical catchup behavior.
--
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]
