anmolxlight opened a new pull request, #70030:
URL: https://github.com/apache/airflow/pull/70030
## Summary
The scheduler drains each executor's event buffer during
`process_executor_events` (via `event_buffer.pop()`) but the database commit
happens later, when the `create_session()` context exits. If the commit raises
a transient database error:
- Callback events have already been removed from the buffer
- The `Callback` row remains in `QUEUED` or `RUNNING` state
- Unlike task instances, executor callbacks have no recovery path that
reselects these states
- The callback can therefore remain stuck indefinitely
## Fix
Save a snapshot of each executor's event buffer before processing. If the
session commit raises an exception, restore the buffer from the snapshot so
events can be retried on the next scheduler loop iteration.
```
# Before: buffer drained inline, commit happens after
with create_session() as session:
for executor in executors:
process_executor_events(executor=executor, session=session)
# After: snapshot, process, restore on commit failure
_event_buffer_snapshots = {}
try:
with create_session() as session:
for executor in executors:
_event_buffer_snapshots[executor] =
executor.get_event_buffer().copy()
for executor in executors:
process_executor_events(executor=executor, session=session)
except Exception:
for executor, snapshot in _event_buffer_snapshots.items():
executor.get_event_buffer().update(snapshot)
raise
```
This is a minimal change at the caller level — the method signature of
`process_executor_events` is unchanged.
## Related issues
Fixes #69975
## Testing
- Existing tests pass
- ruff check and ruff format pass
--
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]
