hkc-8010 commented on code in PR #69821:
URL: https://github.com/apache/airflow/pull/69821#discussion_r3699886887
##########
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:
I changed the retry-eligible trigger-failure path to archive the finished
try explicitly with `prepare_db_for_next_try()` before setting the TI to
`UP_FOR_RETRY`, instead of routing through `handle_failure()` from `DEFERRED`.
That preserves the prior-try history row and keeps log lookup working for try 1.
##########
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:
I removed the direct assignment onto `pm.hook.on_task_instance_failed` and
reworked the regression test so it only asserts the behavior this trigger path
actually guarantees: callback routing plus `task_instance_history` archival on
retry. That avoids leaking a mocked hook into later tests.
--
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]
