Hi all, Raising something that has been coming up over the past few years, especially recently with Asset Watchers: When a Dag is asset-triggered, the scheduler consumes every pending AssetEvent for that Dag in one loop and creates a single DagRun for all of them (_create_dag_runs_asset_triggered). If five files land and generate five events before the next scheduler loop runs, you get one DagRun that consumes all five events, not five separate DagRuns. Most users assume 1 event -> 1 DagRun and read the batchings as Airflow dropping or missing events, and how much gets batched depends on scheduler loop and Dag parallelism settings, not anything declared in the Dag (see https://github.com/apache/airflow/issues/56750). This behavior is intentional, dating back to when datasets shipped in 2.4, but the perception that "this is a bug" is real. This issue has come up as a specific point at Airflow Summit talks two years running, with speakers assuming this was Airflow 2 flakiness that had since been fixed.
It's not just a perception problem, either. Any asset-triggered Dag that isn't explicitly written to iterate `dag_run.consumed_asset_events` will silently under-process when multiple events land in the same run. It looks like "1 run = 1 file" and quietly drops the rest. This isn't a hypothetical scenario. The exact use-case that prompted this is that someone in one of our client facing teams is building an S3 trigger on the AssetWatcher framework (there's no official S3 event trigger for this yet, the only documented AssetWatcher pattern today is SQS) that emits one event per updated file, with several files landing in the same scheduler loop. Because how many events get bundled depends on scheduler cadence and parallelism rather than anything declared in the Dag, this behavior is also untestable. There's no way to write a CI test that reliably asserts "N events produce N runs". There's already community momentum here: - #55956 <https://github.com/apache/airflow/issues/55956> proposes a `max_asset_events` param, milestoned for 3.4.0 - #56750 <https://medium.com/@MarinAgli1/a-look-into-airflow-data-aware-scheduling-and-dynamic-task-mapping-8c548d4ad79> groups the related issues (#53896 <https://github.com/apache/airflow/issues/53896>, #56691 <https://github.com/apache/airflow/issues/56691>, #56050 <https://github.com/apache/airflow/issues/56050> and #47398 <https://github.com/apache/airflow/issues/47398>) and proposes a Dag-level toggle (`asset_grouping`) rather than an all-or-nothing global switch. Regarding the question of whether this can only be a global config: I don't think so. `_create_dag_runs_asset_triggered` already loops per-Dag and only fetches that Dag's own pending events before building its DagRun, so a per-Dag opt-in is localized to that branch and shouldn't require touching the shared code path every other asset-scheduled Dag depends on. I'd model this the same way we already handle `catchup`: a Dag-level parameter (like `asset_grouping`) that falls back to a global `asset_grouping_by_default` in airflow configs when unset, mirroring `catchup` / `catchup_by_default`. That gives Dag authors an explicit override where usage is genuinely mixed, while still letting an org flip the behaviour fleet-wide for every Dag that hasn't opted in, without touching Dag code. SCrocky's issue for example Idescribes running both patterns side by side in the same deployment. Where I want actual discussion: What should `asset_grouping_by_default` ship as? Every proposal so far (including ours) assumes it has to default to today's batched behaviour, to avoid a breaking change. I want to make the case for defaulting it to `False` (1 event -> 1 DagRun) instead, because the cost asymmetric. - If we default to unbatched and someone was relying on batching, the worst case is an extra DagRun. They were never able to control the batch size to begin with as it depends on the scheduler cadence and parallelism, not anything declared in the Dag, so their code already has to tolerate a variable number of events per run. An occasional extra run is just more of the same variance they already had to handle, not a new failure mode. - If we keep batching as the default, everyone who wants per-event semantics, which judging by the summit and this thread is most people's mental model, has to actively work around it. They need to inspect how many events landed in a run and add conditional/expansion logic to split them back out, which is exactly the workaround we're discussing internally right now. This imposes real, ongoing complexity on the majority to protect a minority's default. I don't think this is a close call. I'm not trying to let perfect be the enemy of good. If we can only get consensus on an opt-in with today's default, that's still a real improvement, but I'd rather we make the case for the better default before settling for that. Curious what others thing, especially anyone closer to #56750 <https://medium.com/@MarinAgli1/a-look-into-airflow-data-aware-scheduling-and-dynamic-task-mapping-8c548d4ad79> . Thanks, Constance -- Constance Martineau Staff Product Manager Email: [email protected] Time zone: US Eastern (EST UTC-5 / EDT UTC-4)
