apache / apache/airflow

Idempotency for Event Scheduled DAGs

Open
#49,900 3 comments 0 reactions 0 assignees View on GitHub
affected_version:3.0 kind:feature needs-triage
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.