seanghaeli opened a new issue, #75352:
URL: https://github.com/apache/airflow/issues/75352
### Apache Airflow version
main (development)
### What happened
When two assets declare a watcher with the same trigger class and arguments,
the Dag processor stores a single trigger linked to both assets. The triggerer
then hands that trigger an asset state store covering every linked asset, and
the store's `get`/`set` shortcuts (and `aget`/`aset`) refuse when there is more
than one:
```
ValueError: Task has 2 concrete inlets and outlets — use
context['asset_state_store'][MY_ASSET] to specify which
```
The trigger can't follow that advice either: it isn't told which assets it
watches, and the accessor has no public way to list them. So a shared watcher
trigger has no supported way to persist state.
Today this means:
- `KinesisTrigger` and `IcebergTableSnapshotTrigger` catch the `ValueError`
and keep their cursor in memory only, so for a shared trigger every triggerer
restart re-reads from the initial position (Kinesis) or re-emits the current
snapshot (Iceberg).
- A trigger that doesn't catch it crashes on its first read and is restarted
every couple of seconds without ever firing. That's what `S3KeyUpdateTrigger`
in #74293 does as of now.
- Every new watcher that wants restart-safe state has to copy the same
workaround.
### What you think should happen instead
A watcher trigger should be able to persist state whether it serves one
asset or several.
### How to reproduce
```python
from airflow.providers.amazon.aws.triggers.s3 import S3KeyUpdateTrigger #
any watcher trigger that calls store.get
from airflow.sdk import DAG, Asset, AssetWatcher
def t():
return S3KeyUpdateTrigger(bucket_name="b", bucket_key="k.txt")
x = Asset("asset_x", watchers=[AssetWatcher(name="wx", trigger=t())])
y = Asset("asset_y", watchers=[AssetWatcher(name="wy", trigger=t())])
with DAG("two_assets", schedule=[x, y]):
...
```
There is one `trigger` row linked to `asset_x` and `asset_y`, and
`store.get(...)` raises on the trigger's first step.
### Anything else
Where it comes from:
- `airflow-core/src/airflow/dag_processing/collection.py`,
`add_asset_trigger_references`: watchers are deduplicated by
`BaseEventTrigger.hash(classpath, kwargs)`.
- `airflow-core/src/airflow/jobs/triggerer_job_runner.py`: the trigger gets
an accessor built from every asset linked to it.
- `task-sdk/src/airflow/sdk/execution_time/context.py`:
`AssetStateStoreAccessors._single_accessor` raises when there is more than one.
Some possible directions:
1. Give event triggers a store that works when shared, e.g. `set` writes to
every watched asset and `get` returns the value only if they all agree.
2. Let the trigger see which assets it watches, so it can pick.
3. Don't share a trigger across assets for watchers that use the state store.
At the very least, documenting this for trigger authors would stop new
triggers from crash-looping.
### Are you willing to submit PR?
- [X] Yes I am willing to submit a PR!
---
Drafted-by: Claude Code (Opus 5.5)
--
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]
