Hi all, I've reviewed this PR and the approach looks reasonable to me. A few notes from my review:
- The implementation detects gaps at DagRun creation time and passes a skipped_range DataInterval to the callback/listener. It does not enumerate individual intervals on the scheduler hot path; callers who need that can use `iter_dagrun_infos_between()` in their callback. - The callback is gated: it only runs when catchup=False AND a callback or listener is registered, so there's no overhead for Dags that don't opt in. - The callback follows the existing pipeline (CallbackRequest -> DatabaseCallbackSink -> DagFileProcessor), and the listener fires synchronously, the same as on_dag_run_running. - My review comments have all been addressed. To answer Teghveer's questions: 1. I think this is the right layer for it. The scheduler is already where the "skip" decision happens for catchup=False, so detecting the gap there and emitting a signal seems liek the right place. 2. I don't think a full AIP is warranted. It's a single new constructor parameter (with a listener) and it follows the same pattern as the existing callbacks. The scope is narrow and well-contained. @uranusjr - you requested this discussion; do you have specific concerns about the approach, or can we move forward with review? The PR has been open for two weeks and I feel like it is in good shape from a code perspective. If TP is alright with moving forward, a second review from someone would help get Teghveer's first contribution merged. - ferruzzi ________________________________ From: Teghveer Singh <[email protected]> Sent: Monday, June 22, 2026 8:26 PM To: [email protected] <[email protected]> Subject: [EXT] [DISCUSS] Add on_skipped_intervals_callback / on_intervals_skipped for catchup=False DAGs (PR #68359) CAUTION: This email originated from outside of the organization. Do not click links or open attachments unless you can confirm the sender and know the content is safe. AVERTISSEMENT: Ce courrier électronique provient d’un expéditeur externe. Ne cliquez sur aucun lien et n’ouvrez aucune pièce jointe si vous ne pouvez pas confirmer l’identité de l’expéditeur et si vous n’êtes pas certain que le contenu ne présente aucun risque. Hi all, I'd like to raise a discussion on a new observability feature I've opened in PR #68359 and wanted to get maintainer input before it moves further. *Problem* When a DAG runs with `catchup=False` and the scheduler advances past missed data intervals (e.g., after a restart or re-enable), those skips are silent. No DagRun rows are created, no hook fires. Users who need visibility into which intervals were skipped have no in-process signal and must compare `last_automated_data_interval` against wall clock externally. This was raised in issue #66791. *Proposed Solution* The PR adds two symmetric extension points: 1. `on_skipped_intervals_callback` -- a DAG-level callback on the SDK `DAG` constructor, using the same context-based signature as `on_success_callback` / `on_failure_callback`. The context includes `dag`, `reason` ("skipped_intervals"), and `skipped_range` (a `DataInterval` from the previous automated run's `data_interval_end` to the new run's `data_interval_start`). 2. `on_intervals_skipped` -- an AIP-61 listener hookspec for plugins that need in-process observability without reloading the DAG file. Skipped intervals are detected at scheduled DagRun creation time by comparing the new run's `data_interval_start` against the prior run's `data_interval_end`. The callback/listener only fires when `not dag.catchup` AND a callback or listener is registered, so there is no overhead for DAGs that do not opt in. The callback is dispatched via `DagSkippedIntervalsCallbackRequest` -> `DatabaseCallbackSink` -> DAG File Processor. The listener fires synchronously in the scheduler. Tests are passing locally covering scheduler detection, callback routing and execution, listener firing, and serialization roundtrip. *Questions for the Community:* - Is this the right approach for this observability gap, or is there a preferred pattern? - Should this go through a formal AIP given it introduces a new DAG constructor parameter and listener hookspec? PR: https://github.com/apache/airflow/pull/68359 Original issue: https://github.com/apache/airflow/issues/66791 Happy to address any questions or adjust the approach based on feedback. Best, Teghveer Singh Ateliey
