seanmuth commented on code in PR #73142:
URL: https://github.com/apache/airflow/pull/73142#discussion_r4011271851
##########
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -1768,23 +1768,27 @@ def final_state(self):
"""
The final state of the TaskInstance.
- By default, this will be derived from the exit code of the task
- (0=success, failed otherwise) but can be changed by the subprocess
- sending a TaskState message, as long as the process exits with 0
+ If the subprocess reported a terminal state via message (TaskState,
SucceedTask,
+ RetryTask, etc.) before exiting, that message is authoritative and
takes precedence
+ over the exit code, even if `_check_subprocess_exit` hasn't observed
the process's
+ real exit code yet (`wait()` defaults an unobserved exit code to 1,
which must not be
Review Comment:
You're right — I mis-traced the control flow. Both the `while` condition and
its only `break` require `self._exit_code is not None` first, so the loop can't
exit with the exit code unobserved; that path is dead. Updated the docstring to
describe the reachable scenario instead: a genuinely non-zero exit code
observed *after* a terminal state was already set (e.g.
`_handle_process_overtime_if_needed()`'s SIGTERM), which is what actually needs
this precedence.
---
Drafted-by: Claude Sonnet 5 (no human review before posting)
##########
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -1768,23 +1768,27 @@ def final_state(self):
"""
The final state of the TaskInstance.
- By default, this will be derived from the exit code of the task
- (0=success, failed otherwise) but can be changed by the subprocess
- sending a TaskState message, as long as the process exits with 0
+ If the subprocess reported a terminal state via message (TaskState,
SucceedTask,
+ RetryTask, etc.) before exiting, that message is authoritative and
takes precedence
+ over the exit code, even if `_check_subprocess_exit` hasn't observed
the process's
+ real exit code yet (`wait()` defaults an unobserved exit code to 1,
which must not be
+ allowed to override an already-confirmed terminal state). Only fall
back to deriving
+ the state from the exit code when no terminal message was ever
received.
Not valid before the process has finished.
"""
+ if self._terminal_state is not None:
Review Comment:
Agreed, and updated the docstring to say so explicitly — this now overrides
any observed non-zero exit code once a terminal message arrived, not just an
unobserved one. That's the correct behavior (the message-driven update already
wrote the row; anything else just 409s), but it's worth being upfront that it's
a deliberate precedence change, not only a fix for a dead branch.
---
Drafted-by: Claude Sonnet 5 (no human review before posting)
##########
task-sdk/tests/task_sdk/execution_time/test_supervisor.py:
##########
@@ -4177,6 +4177,29 @@ def
test_non_signal_exit_code_without_retry_goes_to_failed(self, mocker):
assert mock_watched_subprocess.final_state == TaskInstanceState.FAILED
+ def
test_confirmed_terminal_state_takes_precedence_over_unobserved_exit_code(self,
mocker):
+ """
+ A terminal state reported via message (e.g. SucceedTask) is
authoritative even if
+ `wait()` never observed the subprocess's real exit code and defaulted
it to 1 (see
+ `_monitor_subprocess`/`wait()` -- `_check_subprocess_exit` can lose
the race against
+ socket closure under scheduling delay). Regression test for
+ https://github.com/apache/airflow/issues/65708: previously this
returned UP_FOR_RETRY,
+ which caused a spurious `.finish()` call and a 409 against the
already-correct DB row.
Review Comment:
Good catch — the test asserted the new value but didn't actually exercise
the crash. Parametrized over both `should_retry` values and added
`client.task_instances.finish.assert_not_called()`, so `should_retry=False`
(the value that actually reproduces the original 409) is pinned, not just the
value that happens to match the new `final_state`.
---
Drafted-by: Claude Sonnet 5 (no human review before posting)
##########
airflow-core/newsfragments/73142.bugfix.rst:
##########
@@ -0,0 +1 @@
+Fixed a race where a task instance already confirmed ``success`` by the worker
could still be redundantly marked ``up_for_retry`` due to an unobserved
subprocess exit code overriding the already-reported terminal state.
Review Comment:
Removed, thanks.
---
Drafted-by: Claude Sonnet 5 (no human review before posting)
--
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]
