Idempotency for Event Scheduled DAGs
- Dominant language
- Python
- Stars
- 46.9k
- Forks
- 17.8k
- Avg merge
- 2d 10h
- Merged PRs (30d)
- 483
Description
### Description
There should be an option on the AssetWatcher to respect idempotency in the payload of a TriggerEvent could include when yielded, so that a new DAG is only created when a new key shows up. This is already the case for task sensors for HA purposes, so the DAG creation logic surely could respect that too.
This way you could make S3KeyTrigger and similar triggers work by having them put the name of the key that was detected (or modified time if the key is fully defined by the trigger) and put that in the TriggerEvent payload. Then only one DAG would run per time the condition became true, rather than scheduling infinitely.
### Use case/motivation
I was looking to turn a DAG into one with event-driven scheduling and the [notice on infinite scheduling](https://airflow.apache.org/docs/apache-airflow/stable/authoring-and-scheduling/event-scheduling.html#avoid-infinite-scheduling) gave me real pause. It seems like the trigger itself needs to consume the message, whereas my expectation is that most data engineers would find it more natural that the trigger would detect the condition and the running of the DAG would cause the condition to stop being true. E.g. trigger detects a file in a location and the DAG does something with the file then deletes it. But instead what happens is that the [trigger has to delete the file](https://github.com/apache/airflow/blob/main/providers/standard/src/airflow/providers/standard/triggers/file.py#L127).
### Related issues
#49857 would also be fixed by this. Let files pile up in a location with the DAG paused, turn the DAG back on, and suddenly you get an event per file.
### Are you willing to submit a PR?
- [ ] Yes I am willing to submit a PR!
### Code of Conduct
- [x] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
Contributor guide
Research direction
Start with the AssetWatcher and TriggerEvent scheduling path, then inspect airflow/providers/standard/triggers/file.py around the S3KeyTrigger behavior described in the issue. Compare this with the task sensor idempotency behavior mentioned in the report. Done should mean repeated events with the same payload key do not create duplicate DAG runs, while distinct keys still schedule independently.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- python
- Domain
- data-engineering
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 35/100