kaxil commented on code in PR #75874:
URL: https://github.com/apache/airflow/pull/75874#discussion_r4516917568
##########
providers/standard/src/airflow/providers/standard/sensors/external_task.py:
##########
@@ -680,6 +688,9 @@ class ExternalTaskMarker(EmptyOperator):
ui_color = "#4db7db"
operator_extra_links = [ExternalDagLink()]
+ # Where Airflow can follow a recorded clear target, the marker runs on a
worker so that its templates
+ # render there and the resolved target gets recorded, instead of being
marked successful unrun.
+ inherits_from_empty_operator = ExternalClearTarget is None
Review Comment:
Done in 397243fa606. A marker whose IDs have no Jinja, and whose
`logical_date` is unset or a literal, stays an empty operator, and its target
is stored with the Dag version. Core reads it from there and prefers it over a
reported one, so only templated markers run on a worker now. The check uses the
Dag's own Jinja env, so custom delimiters count, and anything that isn't a
string (a callable, `literal()`) still goes to the worker.
##########
airflow-core/src/airflow/api_fastapi/core_api/services/public/task_instances.py:
##########
@@ -593,6 +603,232 @@ def _patch_task_instance_note(
ti.task_instance_note.user_id = user.get_id()
+EXTERNAL_CLEAR_TARGET_XCOM_KEY = "_external_clear_target"
+"""XCom key the Task SDK's task runner records
``ExternalClearTarget.serialize()`` under."""
+
+
+class ExternalClearDepthExceeded(AirflowException):
+ """A clear followed more levels of cross-Dag clear targets than the
starting task allows."""
+
+
+class _RecordedClearTarget(BaseModel):
+ """A clear target as a worker recorded it. Validated, since any task can
write the XCom key."""
+
+ model_config = ConfigDict(frozen=True)
+
+ dag_id: str
+ task_id: str
+ logical_date: AwareDatetime
+ recursion_depth: PositiveInt
+
+
+@dataclass(frozen=True)
+class DependentClear:
+ """What clearing pulls in from one run of another Dag."""
+
+ dag: SerializedDAG
+ run_id: str
+ task_ids: frozenset[str]
+ exclude_task_ids: frozenset[tuple[str, int]]
+ task_instances: list[TI]
+
+
+def _get_recorded_targets(ti_ids: Iterable[UUID], *, session: Session) ->
list[_RecordedClearTarget]:
+ read = build_xcom_read_query(
+ producer_ids=select(TI.id).where(TI.id.in_(ti_ids)),
key=EXTERNAL_CLEAR_TARGET_XCOM_KEY
Review Comment:
Done. If the live attempt hasn't succeeded yet, core uses the latest report
from an earlier attempt of the same execution. Once it succeeds, only its own
report counts, so a rerun that reports nothing doesn't keep the old target.
`test_second_clear_before_the_marker_reruns_still_clears_dependent_dag_tasks`
covers it through the route. Untemplated markers don't hit this at all now,
since their target lives in the Dag version.
##########
task-sdk/src/airflow/sdk/execution_time/task_runner.py:
##########
@@ -2485,6 +2487,17 @@ def _render_map_index(context: Context, ti:
RuntimeTaskInstance, log: Logger) ->
return rendered_map_index
+def _record_external_clear_target(context: Context, ti: RuntimeTaskInstance,
log: Logger) -> None:
+ """Record the task's resolved cross-Dag clear target, so the API server
can follow it on clear."""
+ if not isinstance(ti.task, SupportsExternalClearTarget):
+ return
+ if (target := ti.task.get_external_clear_target(context)) is None:
+ return
+ log.debug("Recording external clear target", target=target)
+ # Straight to the metadata DB, like operator extra links: the API server
reads it there.
+ _xcom_push_to_db(ti, key=EXTERNAL_CLEAR_TARGET_XCOM_KEY,
value=target.serialize())
Review Comment:
Done. The runner now sends the target with the success report
(`external_clear_target` on `TISuccessStatePayload`, added in the 2027-02-28
version), and the API server validates it and writes the record itself. Both
the Execution API and the public XCom API reject the key, and
`exclude_internal_keys` hides it.
##########
airflow-core/src/airflow/api_fastapi/core_api/services/public/task_instances.py:
##########
@@ -593,6 +603,232 @@ def _patch_task_instance_note(
ti.task_instance_note.user_id = user.get_id()
+EXTERNAL_CLEAR_TARGET_XCOM_KEY = "_external_clear_target"
+"""XCom key the Task SDK's task runner records
``ExternalClearTarget.serialize()`` under."""
Review Comment:
Gone with the change above, only core knows the key now
(`airflow.models.xcom`).
--
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]
