KidAmnesiac1 commented on code in PR #74446:
URL: https://github.com/apache/airflow/pull/74446#discussion_r4267086640
##########
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:
Thanks for the feedback @fat-catTW. Can you review the latest change?
--
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]
