sidshas03 commented on code in PR #63355:
URL: https://github.com/apache/airflow/pull/63355#discussion_r2925096943
##########
task-sdk/src/airflow/sdk/execution_time/task_runner.py:
##########
@@ -1349,7 +1360,17 @@ def _on_term(signum, frame):
Stats.incr("ti.finish", tags={**stats_tags, "state": state.value})
if msg:
- SUPERVISOR_COMMS.send(msg=msg)
+ try:
+ SUPERVISOR_COMMS.send(msg=msg)
+ except AirflowRuntimeError as e:
+ is_duplicate, previous_state =
_is_duplicate_state_update(error=e, requested_state=state)
+ if not is_duplicate:
+ raise
+ log.warning(
+ "Task instance state was already updated by API server;
ignoring duplicate state update",
+ requested_state=state.value,
+ previous_state=previous_state,
+ )
Review Comment:
Thanks @amoghrajesh, makes sense. Agree this should be handled in API layer,
not in task runner.
I’ll move this to the TI state endpoint:
- if requested state is same as current state, treat as idempotent no-op
(200)
- keep 409 for real conflicting transitions
I’ll remove the task-runner-side catch from this PR and add API tests for:
1) duplicate same-state update -> success/no-op
2) conflicting state update -> 409
Will push update shortly.
--
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]
