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]

Reply via email to