ataulmujeeb-cyber opened a new pull request, #65240:
URL: https://github.com/apache/airflow/pull/65240
## Summary
Fixes #61451
In Airflow 2, clearing a task upstream of an `ExternalTaskMarker` (with
"Recursive" or "Downstream" selected) would recursively clear the corresponding
`ExternalTaskSensor` and its downstream tasks in the child DAG. This behavior
was lost during the Airflow 2→3 migration.
### Root Cause
In Airflow 2, `DAG.clear()` accepted a `dag_bag` parameter which enabled
cross-DAG clearing traversal. In Airflow 3:
- The `dag_bag` parameter was removed from `SerializedDAG.clear()`
- `clear_task_instances()` has no cross-DAG logic
- The API endpoint handles upstream/downstream only within a single DAG
- All existing tests for this behavior are skipped with
`@pytest.mark.skipif(AIRFLOW_V_3_0_PLUS)`
### Fix
This PR restores cross-DAG recursive clearing by:
1. **New helper function** `clear_external_task_marker_dependencies()` in
`airflow-core/src/airflow/models/taskinstance.py`:
- Detects `ExternalTaskMarker` tasks among cleared task instances (via
serialized `task_type`)
- Loads the referenced external DAGs via `DBDagBag`
- Clears the target tasks and their downstream dependencies in the
external DAG
- Recursively follows transitive `ExternalTaskMarker` chains
- Handles cycle detection via a `visited` set
- Respects the `recursion_depth` parameter to prevent infinite loops
- Gracefully handles missing DAGs/tasks with warnings
2. **New `include_recursive` flag** on `ClearTaskInstancesBody` API model
(`airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py`)
3. **Wired into API endpoint** `post_clear_task_instances()`
(`airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py`)
— calls the new helper when `include_recursive=True`
### Changes
| File | Change |
|---|---|
| `airflow-core/src/airflow/models/taskinstance.py` | Added
`clear_external_task_marker_dependencies()` helper |
|
`airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_instances.py` |
Added `include_recursive` field to `ClearTaskInstancesBody` |
|
`airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py`
| Wired recursive clearing into the clear endpoint |
| `airflow-core/tests/unit/models/test_taskinstance.py` | Added
`TestClearExternalTaskMarkerDependencies` test class |
### Tests
Added 5 tests covering:
- Cross-DAG clearing of external tasks via `ExternalTaskMarker`
- Skipping non-`ExternalTaskMarker` tasks
- Graceful handling of missing external DAGs
- `recursion_depth` limit enforcement (raises `RuntimeError`)
- Cycle detection via visited set
### Notes
- Uses standard Python exceptions (`RuntimeError`) per contributing
guidelines — not `AirflowException`
- The existing Airflow 2 tests for `ExternalTaskMarker` clearing
(`test_external_task_marker_transitive`, etc.) remain skipped for v3 — they
test the old `DagBag`-based clearing path. Once this PR is merged, those could
be updated to test via the API endpoint with `include_recursive=True`.
- The UI would need to expose the `include_recursive` option (similar to the
"Recursive" checkbox in Airflow 2) — that is a follow-up frontend change.
### How to reproduce the original bug
```python
# Parent DAG with ExternalTaskMarker
parent_trigger = ExternalTaskMarker(
task_id='parent_trigger',
external_dag_id='child_dag',
external_task_id='receive_call'
)
# Child DAG with ExternalTaskSensor
receive_call = ExternalTaskSensor(
task_id='receive_call',
external_dag_id='parent_dag',
external_task_id='parent_trigger'
)
```
Before this fix: clearing `parent_trigger` with downstream does NOT clear
`receive_call` in the child DAG.
After this fix: clearing with `include_recursive=True` correctly clears
`receive_call` and its downstream tasks.
### Are you willing to submit PR?
- [x] 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)
--
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]
