anmolxlight commented on code in PR #70030:
URL: https://github.com/apache/airflow/pull/70030#discussion_r4014257183
##########
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:
You are right, and I traced it to confirm: `create_session` commits on exit
and re-raises, my `except` restores the buffer and re-raises,
`_run_scheduler_loop` has no handler around that block (only around the
`_process_task_event_logs` block below it), and both `_execute` and `run_job`
in `job.py` log and re-raise. So the scheduler process exits and the restored
in-memory buffer dies with it. A new process starts with an empty buffer, and
since `_enqueue_executor_callbacks` only selects `PENDING` rows, a
`QUEUED`/`RUNNING` callback whose event was already popped has no recovery
path. The snapshot/restore is dead code in production.
I also agree neither of your options fits this patch: not draining until
after commit means restructuring the shared `process_executor_events` (peek
now, drain post-commit) across all callers, and DB-level reselectability needs
a design (timeout/requeue semantics for stuck callbacks). Both are bigger than
a caller-level patch. Same exposure exists in the sibling blocks in the loop
(`_do_scheduling`, deadlines), so hardening only this one block would be
inconsistent while the scheduler keeps its fail-fast philosophy elsewhere.
Closing this PR and leaving #69975 open for the proper fix. Thanks for the
careful review.
--
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]
