uranusjr commented on code in PR #75446:
URL: https://github.com/apache/airflow/pull/75446#discussion_r4397200316
##########
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -1833,6 +1834,28 @@ def update_task_state_if_needed(self):
)
return
+ if self._execution_timed_out:
+ # The supervisor stopped the task, so report the timeout as the
reason, including a retry, which
+ # would otherwise leave the task instance running until the
scheduler finds it. A task that
+ # exits cleanly without reporting after being stopped has not
succeeded either.
+ reason = f"Task timed out after {self._execution_timeout_seconds}
seconds"
+ if self._should_retry:
+ self.client.task_instances.retry(
+ id=self.id,
+ end_date=datetime.now(tz=UTC),
+ rendered_map_index=self._rendered_map_index,
+ retry_reason=reason,
+ )
+ else:
+ self.client.task_instances.finish(
+ id=self.id,
+ state=TerminalStateNonSuccess.FAILED,
+ when=datetime.now(tz=UTC),
+ rendered_map_index=self._rendered_map_index,
+ retry_reason=reason,
+ )
+ return
Review Comment:
This duplicates `_send_terminal_state_msg` but drops its handling of
CONFLICT, which sets SERVER_TERMINATED. If the server already moved the TI on,
the raw `retry()` or `finish()` call raises.
Build a `RetryTask(retry_reason)` or `TaskState(state=FAILED, retry_reason)`
instead, send it through `_send_terminal_state_msg`.
--
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]
