ashb commented on code in PR #73554:
URL: https://github.com/apache/airflow/pull/73554#discussion_r4109182708
##########
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -1722,13 +1735,28 @@ def wait(self) -> int:
# Now at the last possible moment, when all logs and comms with
the subprocess has finished,
# lets upload the remote logs. Run this in a `finally` so the logs
are uploaded even if the
# state update above raised — a failed state update is exactly
when the logs matter most.
+ self._wait_completed = True
self._upload_logs()
return self._exit_code
def update_task_state_if_needed(self):
if self._terminal_state == SERVER_TERMINATED:
self._pending_terminal_state_msg = None
+ try:
+ self.client.task_instances.finish(
+ id=self.id,
+ state=SERVER_TERMINATED,
+ when=datetime.now(tz=timezone.utc),
+ rendered_map_index=self._rendered_map_index,
+ pid=self.pid,
+ )
+ except ServerResponseError as error:
+ if error.response.status_code != HTTPStatus.NOT_FOUND:
Review Comment:
Checking, but I don't think you can get a 409 here. To hit this, the task
must have started, which means it was running on this host. And the API server
won't let it run somewhere else unless it's been reported as not running here
anymore...
--
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]
