amoghrajesh commented on code in PR #46020:
URL: https://github.com/apache/airflow/pull/46020#discussion_r1931595038
##########
task_sdk/src/airflow/sdk/execution_time/task_runner.py:
##########
@@ -501,6 +503,22 @@ def run(ti: RuntimeTaskInstance, log: Logger):
# TODO: Get a real context object
ti.hostname = get_hostname()
ti.task = ti.task.prepare_for_execution()
+ if ti.task.inlets or ti.task.outlets:
+ inlets = [
+ AssetProfile(name=x.name or None, uri=x.uri or None,
asset_type=Asset.__name__)
+ for x in ti.task.inlets
+ if isinstance(x, Asset)
+ ]
+ outlets = [
+ AssetProfile(name=x.name or None, uri=x.uri or None,
asset_type=Asset.__name__)
+ for x in ti.task.outlets
+ if isinstance(x, Asset)
+ ]
+ SUPERVISOR_COMMS.send_request(msg=RuntimeCheckOnTask(inlet=inlets,
outlet=outlets), log=log) # type: ignore
+ msg = SUPERVISOR_COMMS.get_message() # type: ignore
+ if isinstance(msg, OKResponse) and not msg.ok:
+ log.info("Runtime checks failed for task, marking task as
failed..")
+ return
Review Comment:
The message isn't send. It comes from the supervisor from line 518. And if
its a "not ok", the task runner logs and doesn't perform rest of the code.
The logic behind OKResponse is like this:
1. Task runner sends a `RuntimeCheckOnTask` call to supervisor
2. Supervisor uses http client to make a REST api call to `runtime_checks`
endpoint in the execution API.
3. Supervisor gets `OKResponse` from client
4. Supervisor sends it down to task runner.
5. On line 519, task runner checks if everything is OK, if not, it exits and
doesn't execute rest of the code.
--
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]
