fat-catTW commented on code in PR #76303:
URL: https://github.com/apache/airflow/pull/76303#discussion_r4453065235
##########
providers/amazon/src/airflow/providers/amazon/aws/executors/ecs/ecs_executor.py:
##########
Review Comment:
Should this preserve the task's queue?
Passing `queue=None` sends routed callbacks to the default ECS queue, rather
than the task's configured queue.
##########
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:
I think this might need cross-scheduler coordination.
In HA, each scheduler can calculate half of the same free capacity and
collectively fill all slots with callbacks.
--
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]
