[v3-3-test] Fix task callbacks being skipped when TriggerDagRunOperator gets a 404 (#70719) - #71083
Merged
Merged
Conversation
…tor` gets a 404 (#70719) * Fix task callbacks being skipped when TriggerDagRunOperator gets a 404 `run()` maps a task's outcome through a flat chain of `except` clauses, and several of those clauses do real work: they call the API server, or serialize user-supplied values. Python does not offer an exception raised inside an `except` clause to that clause's siblings, so when one of them raised, the exception escaped `run()` entirely -- skipping the retry decision in `_handle_current_task_failed()` and every callback, listener and failure email in `finalize()`. The reported path: triggering a Dag that does not exist returns 404, which `DagRunOperations.trigger` re-raises (it only special-cases the 409 already-exists case). The supervisor turns it into an `API_SERVER_ERROR` response and `CommsDecoder._from_frame` raises `AirflowRuntimeError` -- from inside `except DagRunTriggerException`, a few lines above the clause that already handles `AirflowRuntimeError`. It is not limited to that path: `_defer_task` and `_await_input_task` run `serde_serialize` over user-supplied kwargs, which raises `TypeError` for any value serde has no serializer for. Split the function at the point where deciding the outcome ends and reporting it begins. `_run_task_and_map_outcome()` keeps the chain verbatim and returns the outcome; `run()` calls it, and its `except` now covers the handlers too. The chain and the terminal-state `finally` block are untouched, so this is a behaviour change rather than a reshuffle of existing lines. `_handle_handler_failure()` keeps the chain's own classifications instead of routing everything through the retry-count check, so `AirflowFailException`, `AirflowSensorTimeout` and `AirflowTaskTerminated` still fail without retrying when they surface from a handler. It catches `Exception`, not `BaseException`, so `KeyboardInterrupt` still reaches `main()`'s exit-code-2 path -- the supervisor's default termination signal is SIGINT, so swallowing it would turn an operator-initiated kill into an ordinary retry. If the failure path itself raises, it fails closed on a plain FAILED state rather than re-entering the code that just failed and escaping again. * Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --------- (cherry picked from commit f0c13dc) Co-authored-by: Kaxil Naik <kaxilnaik@gmail.com> Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
run()maps a task's outcome through a flat chain ofexceptclauses, andseveral of those clauses do real work: they call the API server, or serialize
user-supplied values. Python does not offer an exception raised inside an
exceptclause to that clause's siblings, so when one of them raised, theexception escaped
run()entirely -- skipping the retry decision in_handle_current_task_failed()and every callback, listener and failure emailin
finalize().The reported path: triggering a Dag that does not exist returns 404, which
DagRunOperations.triggerre-raises (it only special-cases the 409already-exists case). The supervisor turns it into an
API_SERVER_ERRORresponse and
CommsDecoder._from_frameraisesAirflowRuntimeError-- frominside
except DagRunTriggerException, a few lines above the clause thatalready handles
AirflowRuntimeError. It is not limited to that path:_defer_taskand_await_input_taskrunserde_serializeover user-suppliedkwargs, which raises
TypeErrorfor any value serde has no serializer for.Split the function at the point where deciding the outcome ends and reporting
it begins.
_run_task_and_map_outcome()keeps the chain verbatim and returnsthe outcome;
run()calls it, and itsexceptnow covers the handlers too.The chain and the terminal-state
finallyblock are untouched, so this is abehaviour change rather than a reshuffle of existing lines.
_handle_handler_failure()keeps the chain's own classifications instead ofrouting everything through the retry-count check, so
AirflowFailException,AirflowSensorTimeoutandAirflowTaskTerminatedstill fail without retryingwhen they surface from a handler. It catches
Exception, notBaseException,so
KeyboardInterruptstill reachesmain()'s exit-code-2 path -- thesupervisor's default termination signal is SIGINT, so swallowing it would turn
an operator-initiated kill into an ordinary retry. If the failure path itself
raises, it fails closed on a plain FAILED state rather than re-entering the
code that just failed and escaping again.
Co-authored-by: Copilot Autofix powered by AI 175728472+Copilot@users.noreply.github.com
(cherry picked from commit f0c13dc)
Co-authored-by: Kaxil Naik kaxilnaik@gmail.com
Co-authored-by: Copilot Autofix powered by AI 175728472+Copilot@users.noreply.github.com