ephraimbuddy commented on code in PR #76303:
URL: https://github.com/apache/airflow/pull/76303#discussion_r4454160899
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -1450,10 +1534,393 @@ def _enqueue_executor_callbacks(self, session:
Session) -> None:
generator=executor.jwt_generator,
)
- executor.queue_workload(workload, session=session)
- callback.state = CallbackState.QUEUED
+ try:
+ executor.queue_workload(workload, session=session)
+ except NotImplementedError as e:
+ # Some executors run callbacks without declaring it, so
ask the executor itself.
+ self.log.error("Failing callback %s: %s", callback.id, e)
+ callback.state = CallbackState.FAILED
+ callback.output = str(e)
+ else:
+ callback.state = CallbackState.QUEUED
session.add(callback)
+ def _route_dag_processor_callbacks(self, *, session: Session) -> None:
+ """
+ Queue Dag processor callbacks on executors.
+
+ With routing on, callback sinks store Dag processor callbacks as
executor callbacks waiting
+ SCHEDULED (see ``ExecutorCallback.from_dag_processor_request``). A
routed callback is marked
+ QUEUED in the transaction that queues it. One that no executor can run
is failed with the
+ reason, and one that fails to route is tried again in a later loop, up
to
+ ``MAX_ROUTED_CALLBACK_TRIES``.
+
+ Callbacks take at most half the free slots globally and on each
executor, but one while any
+ is free, before the loop queues tasks into the rest: executors send
callbacks first, and one
+ left waiting in an executor for a slot would age toward its start
deadline. Callbacks that
+ could be routed but find no free slot wait for a later loop, while
those no executor can run
+ are failed without waiting for one.
+ """
+ free_slots = self._parallelism - sum(executor.slots_occupied for
executor in self.executors)
Review Comment:
The slot counters here are per scheduler: `slots_occupied` counts its own
executor workloads, and `core.parallelism` applies to each scheduler
separately. So they aren’t dividing the same logical capacity. But you’re right
that, with a shared worker pool, this doesn’t guarantee spare worker capacity
for tasks.
“Globally” meant across this scheduler’s executors, but it reads as
cluster-wide. I’ll clarify that this is a per-scheduler, per-loop budget,
including the minimum-one exception, and soften the claim about preventing
callbacks from blocking tasks.
--
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]
