kaxil commented on code in PR #48880:
URL: https://github.com/apache/airflow/pull/48880#discussion_r2031529428
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -323,7 +330,7 @@ def start( # type: ignore[override]
proc = super().start(id=job.id, job=job, target=cls.run_in_process,
logger=logger, **kwargs)
msg = messages.StartTriggerer(requests_fd=proc._requests_fd)
- proc._send(msg)
+ proc.stdin.write(msg.model_dump_json().encode("utf-8") + b"\n")
Review Comment:
```suggestion
proc.stdin.write(msg.model_dump_json().encode() + b"\n")
```
Since `utf-8` is the default
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -908,9 +905,23 @@ async def sync_state_to_supervisor(self, finished_ids:
list[int]):
if not finished_ids:
msg.finished = None
- if msg.events or msg.finished or msg.failures:
+ # Block triggers from making any requests for the duration of this
+ async with SUPERVISOR_COMMS.lock:
# Tell the monitor that we've finished triggers so it can update
things
self.requests_sock.write(msg.model_dump_json(exclude_none=True).encode() +
b"\n")
+ # async for line in stdin:
+ # msg = self.decoder.validate_json(line)
Review Comment:
```suggestion
```
--
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]
