kaxil commented on code in PR #73142:
URL: https://github.com/apache/airflow/pull/73142#discussion_r4011099818
##########
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:
I don't think this race is reachable. `_monitor_subprocess()` loops on
`while self._exit_code is None or self._open_sockets:` and its only `break` is
nested inside `if self._exit_code is not None`, so it can't return with the
exit code still unobserved, which makes the `else 1` fallback in `wait()` dead
on that path.
What is reachable is a subprocess that genuinely exits non-zero after
`succeed()` already returned 204. `_handle_process_overtime_if_needed()` is one
concrete way to get there: it only fires once `_terminal_state` is set, and it
kills with SIGTERM, so the exit code is a real, observed, negative one. The fix
still holds for that case, but the docstring as written sends the next reader
after a race the loop already prevents. Same wording is in the newsfragment and
the test docstring.
##########
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:
This is broader than what the docstring above describes: it returns the
terminal state even when the exit code was genuinely observed and non-zero. A
task that sent `SucceedTask` and then died in teardown (OOM kill, or the
`task_success_overtime` SIGTERM) now reports `success` where it used to report
`up_for_retry` or `failed`. That looks right to me since `succeed()` already
wrote the row and anything else just 409s, but it's a behaviour change in its
own right, not only a fix for an unobserved exit code.
##########
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:
With `_should_retry = True` the old code returned `UP_FOR_RETRY`, which is
in `STATES_SENT_DIRECTLY`, so `update_task_state_if_needed()` would have
skipped `finish()` and no 409 would follow. The crash needs the other branch:
`_should_retry = False`, where the old code fell through to `FAILED`, which
isn't in that set. The assertion is a valid regression on `final_state` either
way, but as written it doesn't cover the case that actually crashed.
Parametrizing `_should_retry` over both values would, and asserting
`client.task_instances.finish` isn't called would pin the user-visible half of
the bug.
--
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]
