seanghaeli opened a new pull request, #75521: URL: https://github.com/apache/airflow/pull/75521
When several assets have a watcher with the same trigger class and arguments, they share one trigger, but the triggerer gives it the task accessor for the asset state store, which refuses `get`/`set` when there's more than one asset. So a shared watcher can't keep state: `S3KeyUpdateTrigger` (#74293) crash-loops, and `KinesisTrigger` and `IcebergTableSnapshotTrigger` fall back to an in-memory position. Every event a shared trigger fires already goes to all of its assets, so it only has one position to keep. Triggers now get their own accessor: `set`/`delete`/`clear` apply to every asset, and `get` returns the value from the first asset (by name) that has the key, so an asset added to the trigger later doesn't cause a re-fire. Tasks keep the current accessor, and a single-asset trigger sends the same calls as before. Tested in Breeze with #74293's `S3KeyUpdateTrigger` on top, two assets x and y sharing one watcher on a moto S3 object: ``` before (main + #74293), c.txt uploaded, 60s: a1: trigger crashes=27 [error] Trigger ID 1 exited with error Task has 2 concrete inlets and outlets — use context['asset_state_store'][MY_ASSET] to specify which (no asset events) after, c.txt uploaded while the triggerer runs: asset_event id=1 asset=e2e75352_x etag="6654c734ccab8f440ff0825eb443dc7f" previous_etag=None asset_event id=2 asset=e2e75352_y etag="6654c734ccab8f440ff0825eb443dc7f" previous_etag=None triggerer restarted, file unchanged: no new events c.txt modified, then re-uploaded with the same content: asset_event id=3 asset=e2e75352_x etag="1b267619c4812cc46ee281747884ca50" previous_etag="6654c734ccab8f440ff0825eb443dc7f" asset_event id=4 asset=e2e75352_y etag="1b267619c4812cc46ee281747884ca50" previous_etag="6654c734ccab8f440ff0825eb443dc7f" triggerer restarted, file unchanged: no new events asset z added to the trigger, restarted, then c.txt modified: trigger id=1 assets=['e2e75352_x', 'e2e75352_y', 'e2e75352_z'] asset_event id=5 asset=e2e75352_x etag="43a03299a3c3fed3d8ce7b820f3aca81" previous_etag="1b267619c4812cc46ee281747884ca50" asset_event id=6 asset=e2e75352_y etag="43a03299a3c3fed3d8ce7b820f3aca81" previous_etag="1b267619c4812cc46ee281747884ca50" asset_event id=7 asset=e2e75352_z etag="43a03299a3c3fed3d8ce7b820f3aca81" previous_etag="1b267619c4812cc46ee281747884ca50" trigger crashes in every run: 0 ``` closes: #75352 --- ##### Was generative AI tooling used to co-author this PR? - [X] Yes (please specify the tool below) Generated-by: Claude Code (Opus 5.5) following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions) -- 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]
