kaxil opened a new pull request, #75874: URL: https://github.com/apache/airflow/pull/75874
Fixes #61451. An alternative to #65314. In Airflow 2, clearing a task with its downstream tasks also cleared the task an `ExternalTaskMarker` points at, and everything downstream of it in the other Dag. Airflow 3 dropped this. The Airflow 2 code rendered the marker's templates inside the webserver, and the Airflow 3 API server cannot run user code. With this change the marker resolves its target on a worker and records it. The API server follows that record when it clears. ## Design rationale **Core does not look for `ExternalTaskMarker` by name.** #65314 does, which ties core to a provider class name and to provider attributes on a serialized operator. That approach also misses templated targets: the marker is a trivial `EmptyOperator`, so the scheduler marks it successful without running it and no rendered-fields row is ever written. With a custom `logical_date` template (an hourly Dag marking its daily child's run, say) or a templated `external_dag_id`, the clear lands on the wrong run or on none. **How it works:** - **Task SDK.** `ExternalClearTarget(dag_id, task_id, logical_date, recursion_depth)`. An operator opts in by defining `get_external_clear_target(context)`; the task runner finds it by name, through a `runtime_checkable` `Protocol`, the way OpenLineage finds `get_openlineage_facets_on_complete`. `BaseOperator` is unchanged. After a successful run the runner stores the result in the metadata DB under the XCom key `_external_clear_target`, the way operator extra links are stored. If the method raises, the task fails, so a bad template surfaces where it was written. - **Core.** `find_external_clear_task_instances` follows the records level by level, capped by `recursion_depth`. It takes the downstream tasks from each target run's own Dag version. Core parses the record itself and does not import the Task SDK at runtime. - **REST API.** `ClearTaskInstancesBody.include_dependent_dags: bool | None`. Left unset, it follows `include_downstream`, so the UI's default clear cascades. `false` turns it off. Task instances in target Dags the caller cannot edit are left out, and each target Dag uses its own `rerun_with_latest_version`. - **Provider.** `ExternalTaskMarker` defines the method. On an Airflow that has the new type, it is no longer treated as an empty operator, so it runs on a worker. **The runner calls the method; `execute()` does not raise or push.** I also tried the `TriggerDagRunOperator` style, where the marker's `execute()` raises a signal for the task runner to catch. There, a subclass whose `execute()` skips `super()` or wraps it in `try/except Exception` drops the cascade, and the marker still shows success. Pushing an XCom from `execute()` has the same `super()` problem, and a custom XCom backend can store the record as a reference the API server cannot read. With the runner calling the method after `execute()` returns, neither applies. **Nothing on `BaseOperator`.** Only one operator needs this today, so it is not a base-class method or a mixin. **Mapped markers.** `OperatorPartial._expand` decided `is_empty` with `issubclass(..., EmptyOperator)`. It ignored a subclass that sets `inherits_from_empty_operator = False`, so mapped markers never ran. It now reads the attribute, the same way an unmapped operator does, which also removes the Task SDK's import of the standard provider's `EmptyOperator`. ## Gotchas - Each marker now uses a worker slot on every run. In a local run, execution took about 2 s for the first marker and under 0.1 s for later ones, not counting queueing. - A marker that ran before this change, or never succeeded, has no record, so clearing it does not cascade. - The depth cap comes from the largest `recursion_depth` among the markers in the first level (Airflow 2 took it from whichever marker it hit first). ## Known issues - The core and Task SDK changes and the standard provider change ship in different releases. The provider only takes the worker path when the new type is importable. - `airflow tasks clear` from the CLI does not cascade yet. -- 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]
