kaxil commented on code in PR #74528:
URL: https://github.com/apache/airflow/pull/74528#discussion_r4501658756
##########
providers/standard/src/airflow/providers/standard/sensors/external_task.py:
##########
@@ -378,16 +392,34 @@ def _get_count(states: list[str]) -> int:
if self.failed_states:
count = _get_count(self.failed_states)
count_failed = self._calculate_count(count, dttm_filter)
+ if count_failed > 0:
Review Comment:
Nothing pins these guards. Making this push unconditional, doing the same
for `count_skipped`, or moving the success push above the `return False` all
keep the test file green, since every V3 test ends the wait on the first poke
and the `ti` from `setup_method` is an unspecced `MagicMock`. Could you add a
V3 case where the counts are 0 and assert `poke()` returns False without
calling `get_task_states` or `xcom_push`?
##########
providers/standard/src/airflow/providers/standard/sensors/external_task.py:
##########
@@ -65,6 +67,13 @@
from airflow.providers.common.compat.sdk import Context, TaskInstanceKey
+XCOM_EXTERNAL_RUN_IDS = "external_run_ids"
+
+
+def _build_airflow_ui_url(path: str) -> str:
Review Comment:
This is core's `build_airflow_dagrun_url` with the path left open. Could
`get_link` use it for the single-run case too
(`f"dags/{external_dag_id}/runs/{run_id}"`) and drop the
`airflow.utils.helpers` import?
##########
providers/standard/src/airflow/providers/standard/sensors/external_task.py:
##########
@@ -378,16 +392,34 @@ def _get_count(states: list[str]) -> int:
if self.failed_states:
count = _get_count(self.failed_states)
count_failed = self._calculate_count(count, dttm_filter)
+ if count_failed > 0:
+ self._push_external_run_ids(context, dttm_filter)
self._handle_failed_states(count_failed)
if self.skipped_states:
count = _get_count(self.skipped_states)
count_skipped = self._calculate_count(count, dttm_filter)
+ if count_skipped > 0:
+ self._push_external_run_ids(context, dttm_filter)
self._handle_skipped_states(count_skipped)
count = _get_count(self.allowed_states)
count_allowed = self._calculate_count(count, dttm_filter)
- return count_allowed == len(dttm_filter)
+ if count_allowed != len(dttm_filter):
+ return False
+ self._push_external_run_ids(context, dttm_filter)
+ return True
+
+ def _push_external_run_ids(self, context: Context, dttm_filter:
Sequence[datetime.datetime]) -> None:
Review Comment:
This runs ahead of the outcome: before `_handle_failed_states` /
`_handle_skipped_states` raise, before `return True`, and before
`execute_complete` reads the event. So if the extra `get_task_states` call or
the XCom push fails, the link decides the task state. A `soft_fail=True` sensor
that hit `failed_states` fails instead of skipping, and a sensor whose upstream
succeeded fails or retries. `finalize()` wraps `get_link` in a `try/except` so
a link can't fail the task. Could this catch `Exception` and log a warning, so
a failed lookup only costs the link? `get_link` already falls back to the Dag
page when the key is missing.
--
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]
