fix(task_runner): treat DAGRUN_ALREADY_EXISTS as idempotent success on transport-error retry - #68828
Closed
kalluripradeep wants to merge 2 commits into
Closed
Conversation
When a trigger POST creates the DAG run server-side but the client observes an httpx.RequestError, the tenacity retry in Client.request() replays the POST. The server returns 409 Conflict, which the client converts to ErrorResponse(DAGRUN_ALREADY_EXISTS). Previously _handle_trigger_dag_run treated this as a fatal conflict and marked the parent task FAILED, without writing the trigger_run_id XCom. The requested run_id already exists -- which is exactly the desired outcome. With skip_when_already_exists=False (the default) there is no user intent to skip or fail on conflict; the DAGRUN_ALREADY_EXISTS is purely an artefact of the transport-error retry path. Fix: when DAGRUN_ALREADY_EXISTS is received and skip_when_already_exists is False, log an info message and fall through to the normal success path (XCom push + wait_for_completion handling). The skip_when_already_exists branch is unchanged -- it still returns SKIPPED immediately. Fixes apache#66905
…sion test For test_handle_trigger_dag_run_conflict: - Change (False, FAILED) to (False, SUCCESS): when skip_when_already_exists=False DAGRUN_ALREADY_EXISTS is now treated as idempotent success. - Update docstring to reflect the new semantics. Add test_handle_trigger_dag_run_transport_retry_idempotent_success: - Verifies that DAGRUN_ALREADY_EXISTS after a transport-error retry produces TaskInstanceState.SUCCESS and pushes the trigger_run_id XCom. - Uses side_effect to simulate the two SUPERVISOR_COMMS.send() calls: TriggerDagRun (409 path) followed by SetXCom. Fixes apache#66905
Contributor
Author
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
1 participant
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.
Summary
Fixes #66905 —
TriggerDagRunOperatorcan fail withDagRunAlreadyExistsafter an ambiguous execution-API trigger retry.Root cause
Client.request()has a tenacity@retrydecorator that replays requests onhttpx.RequestError(transport-level ambiguities) and HTTP 5xx. When aPOST dag-runs/{dag_id}/{run_id}creates the DAG run server-side but the response is lost, the retry receives409 Conflict.DagRunOperations.trigger()converts that toErrorResponse(DAGRUN_ALREADY_EXISTS)and returns it to the supervisor, which forwards it to the task runner.Previously
_handle_trigger_dag_runreacted toDAGRUN_ALREADY_EXISTSwithskip_when_already_exists=False(the default) by marking the parent task FAILED and not writing thetrigger_run_idXCom.Fix
When
DAGRUN_ALREADY_EXISTSis received andskip_when_already_exists=False, the requestedrun_idalready exists — which is the desired end state. Instead of failing, log an info message and fall through to the normal success path: push thetrigger_run_idXCom and honourwait_for_completion.The
skip_when_already_exists=Truebranch is unchanged — it still returnsSKIPPEDimmediately.Changes
task-sdk/src/airflow/sdk/execution_time/task_runner.py_handle_trigger_dag_run: changeelsebranch from FAIL to fall-through-to-successtask-sdk/tests/task_sdk/execution_time/test_task_runner.py(False, FAILED)→(False, SUCCESS); add dedicated transport-retry regression testTesting