fat-catTW commented on code in PR #74446:
URL: https://github.com/apache/airflow/pull/74446#discussion_r4227914569
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -1371,15 +1371,24 @@ def _enqueue_executor_callbacks(self, session: Session)
-> None:
)
continue
- workload = workloads.ExecuteCallback.make(
- callback=callback,
- dag_run=dag_run,
- generator=executor.jwt_generator,
- )
-
- executor.queue_workload(workload, session=session)
- callback.state = CallbackState.QUEUED
- session.add(callback)
+ # A savepoint so that a callback the executor cannot store
(e.g. a DB write rejected at
+ # flush) fails on its own instead of aborting the scheduler
loop. Otherwise the rollback
+ # leaves it PENDING and every scheduler restart crashes on it
again.
+ try:
+ with session.begin_nested():
Review Comment:
I think this needs one more cleanup step for executor-side state.
The savepoint rolls back the DB write, but not executor memory. If
`queue_workload()` already added the workload to `executor.executor_queues`
before the flush fails, we can mark the callback `FAILED` while still leaving
it queued for the next executor heartbeat.
That leaves the system in an inconsistent state:
- DB: callback is `FAILED`
- executor memory: same callback workload is still queued/running
Could we remove the workload from executor tracking in the exception path,
and assert that in the regression test? Something along these lines:
```python
workload = None
try:
with session.begin_nested():
workload = workloads.ExecuteCallback.make(...)
executor.queue_workload(workload, session=session)
callback.state = CallbackState.QUEUED
session.add(callback)
except Exception as e:
if workload is not None:
executor.executor_queues[workload.type].pop(workload.key, None)
executor.running.discard(workload.key)
self.log.exception("Failed to queue callback %s to %s", callback.id,
executor)
callback.state = CallbackState.FAILED
callback.output = f"Failed to queue callback: {type(e).__name__}"
session.add(callback)
--
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]
