kaxil commented on code in PR #69821:
URL: https://github.com/apache/airflow/pull/69821#discussion_r3681509912
##########
airflow-core/src/airflow/models/trigger.py:
##########
@@ -604,10 +633,21 @@ def _submit_callback_if_necessary() -> None:
def _push_xcoms_if_necessary() -> None:
"""Pushes XComs to the database if they are provided."""
- if event.xcoms:
+ if event.xcoms and callback_type != TaskInstanceState.UP_FOR_RETRY:
for key, value in event.xcoms.items():
task_instance.xcom_push(key=key, value=value)
+ # Send the callback before handle_failure (mirrors the scheduler
executor-event ordering):
+ # handle_failure -> save_to_db commits, which also persists the
DatabaseCallbackSink row atomically.
_submit_callback_if_necessary()
+
+ if handle_via_failure:
+ # Canonical failure handling: sets UP_FOR_RETRY/FAILED by
retry-eligibility, fires the
+ # on_task_instance_failed listener + failure metrics + Log audit row,
and clears
+ # next_method args. Mirrors the scheduler executor-event path (PR
#56586).
+ task_instance.handle_failure(error="Task failed via trigger event",
session=session)
Review Comment:
On the retry-eligible path this skips the try archival the worker-side retry
performs: `fetch_handle_failure_context` only calls `prepare_db_for_next_try`
when `ti.state == RUNNING` (taskinstance.py:1870), and here the TI is still
`deferred`, so no `task_instance_history` row is written for try 1 and `ti.id`
is not rotated. When the scheduler then makes the TI schedulable it bumps
`try_number` to 2 (dagrun.py:2144), and `GET
.../taskInstances/{task_id}/logs/1` 404s at log.py:157 because there is no live
TI at try 1 and no history row to fall back on. Worth deciding whether this
branch should archive the try itself, or route through the
`submit_failure`/`__fail__` path that already picks up the worker-side
machinery.
##########
airflow-core/tests/unit/models/test_trigger.py:
##########
@@ -362,6 +366,63 @@ def
test_submit_event_task_end_callback_includes_version_data(mock_send, session
assert request.version_data == version_data
[email protected](
+ ("retries", "expected_state", "expected_callback_type"),
+ [
+ (1, TaskInstanceState.UP_FOR_RETRY, TaskInstanceState.UP_FOR_RETRY),
+ (0, TaskInstanceState.FAILED, TaskInstanceState.FAILED),
+ ],
+)
+@patch("airflow.callbacks.database_callback_sink.DatabaseCallbackSink.send")
+def test_submit_event_task_end_failed_respects_retries(
+ mock_send, session, create_task_instance, retries, expected_state,
expected_callback_type
+):
+ """A trigger-emitted TaskFailedEvent should respect retry-eligibility: a
deferred task with
+ retries remaining goes UP_FOR_RETRY (on_retry_callback), not straight to
FAILED.
+
+ Failures are routed through ``TaskInstance.handle_failure`` (mirroring the
scheduler
+ executor-event path), so the ``on_task_instance_failed`` listener fires in
both cases.
+ """
+ from airflow.listeners.listener import get_listener_manager
+
+ listener_callback = MagicMock()
+ get_listener_manager().pm.hook.on_task_instance_failed = listener_callback
Review Comment:
Assigning onto `pm.hook.on_task_instance_failed` leaks past the `finally`:
`clear()` only unregisters plugins
(shared/listeners/src/airflow_shared/listeners/listener.py:121-124) and
`get_listener_manager()` is `@cache`d, so the MagicMock stays on the shared
hook relay for the rest of the pytest worker. A later test that registers a
real listener then dies in pluggy's `_verify_hook` with `InvalidSpecError:
Cannot spec a Mock object`. The `listener_manager` fixture in
`tests_common/pytest_plugin.py` (or `monkeypatch.setattr`) restores the
original hook.
--
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]
