uranusjr commented on code in PR #40017:
URL: https://github.com/apache/airflow/pull/40017#discussion_r1625746888
##########
airflow/jobs/scheduler_job_runner.py:
##########
@@ -986,11 +1023,23 @@ def _run_scheduler_loop(self) -> None:
self.processor_agent.wait_until_finished()
with create_session() as session:
+ # This will schedule for as many executors as possible.
num_queued_tis = self._do_scheduling(session)
- self.job.executor.heartbeat()
+ # Heartbeat all executors, even if they're not receiving
new tasks this loop. It will be
+ # either a no-op, or they will check-in on currently
running tasks and send out new
+ # events to be processed below.
+ for executor in self.job.executors:
+ executor.heartbeat()
+
+ # TODO: I'm not sure why we expunge here. Do we need to do
it after each individual
Review Comment:
Seemingly random expunges are usually for HA (running multiple scheduler
processes). This seems to be the case; the line was first introduced in
73b9163a8f55ce3d5bf6aec0a558952c27dd1b55. So this is fine—*maybe* we can do
this before heartbeating to avoid some confusion? But that might mess up some
edge cases a bit since expunging might take some time (???) not exactly sure.
--
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]
