amoghrajesh commented on code in PR #45106:
URL: https://github.com/apache/airflow/pull/45106#discussion_r1898452156
##########
task_sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -710,7 +714,14 @@ def handle_requests(self, log: FilteringBoundLogger) ->
Generator[None, bytes, N
def _handle_request(self, msg: ToSupervisor, log: FilteringBoundLogger):
log.debug("Received message from task runner", msg=msg)
resp = None
- if isinstance(msg, TaskState):
+ if isinstance(msg, FailState):
+ self._terminal_state = TerminalTIState.FAILED
+ self._task_end_time_monotonic = time.monotonic()
+ self._fail_request_sent = True
Review Comment:
The reason we have this variable is so that we can maintain a sync between
"handle_requests" and "wait".
Here, we want to send the request to the execution API server with `msg`
that we received from task runner (as there is no real way to share data
between `handle_requests` and `wait`) as sent here:
https://github.com/apache/airflow/pull/45106/files#diff-c2651fdee1a25e091e2a9d4f937f8032ca3d289d0de76f38ed88aee5df0f880dR722.
Once we send this request, we mark the `_fail_request_sent` as True. Which
indicates that the request to `fail` was sent. Now the portion in `wait` will
not send it again as checked here:
https://github.com/apache/airflow/pull/45106/files#diff-c2651fdee1a25e091e2a9d4f937f8032ca3d289d0de76f38ed88aee5df0f880dR528
I added code level comments, should I update that?
--
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]
