steveahnahn commented on code in PR #70030:
URL: https://github.com/apache/airflow/pull/70030#discussion_r3692875597
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -1807,12 +1807,24 @@ def _run_scheduler_loop(self) -> None:
):
executor.heartbeat()
- with create_session() as session:
- num_finished_events = 0
- for executor in self.executors:
- num_finished_events += self._process_executor_events(
- executor=executor, session=session
- )
+ # ponytail: snapshot event_buffer dict directly (not
get_event_buffer)
+ # so mock-based tests that count get_event_buffer calls pass.
+ # Accessing event_buffer by attribute returns the same dict as
+ # the get_event_buffer method on real executors.
+ _event_buffer_before: dict[BaseExecutor, dict] = {}
+ try:
+ with create_session() as session:
+ for executor in self.executors:
+ _event_buffer_before[executor] =
executor.event_buffer.copy()
+ num_finished_events = 0
+ for executor in self.executors:
+ num_finished_events +=
self._process_executor_events(
+ executor=executor, session=session
+ )
+ except Exception:
+ for executor, buf in _event_buffer_before.items():
+ executor.event_buffer.update(buf)
+ raise
Review Comment:
No enclosing handler catches this. The only try in _run_scheduler_loop wraps
_process_task_event_logs further down, and _execute's handler around the loop
call re-raises. So the process exits here.
To survive a transient commit failure you'd probably need to not drain until
after commit, or make the stuck Callback rows reselectable so recovery comes
from the DB. Would either fit the case you hit?
--
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]
