Auto Routing trigger instances to different triggerer

[HuanjieGuo]

Currently, the capacity is set to default 1000 in TriggererJobRunner.

https://github.com/apache/airflow/blob/main/airflow/jobs/triggerer_job_runner.py#L264

When a batch of tasks are triggered at the same time, these trigger jobs may be sent to only one triggerer and other would do nothing.

It would be better if we write a logic here to calculate how many trigger jobs a triggerer should get from DB here and let other triggerers have a chance to dispatch the trigger jobs.

https://github.com/apache/airflow/blob/main/airflow/jobs/triggerer_job_runner.py#L398

My initial idea is to change it like this

    def load_triggers(self):
        recommend_capacity = total_trigger_job_cnt / active_triggerer_count + 1
        instance_capacity = math.min(self.capacity, recommend_capacity)
        Trigger.assign_unassigned(self.job.id, instance_capacity, self.health_check_threshold)
        ids = Trigger.ids_for_triggerer(self.job.id)
        self.trigger_runner.update_triggers(set(ids))

@potiuk is this a point to contribute? I want to submit a PR for it.

[potiuk]

@potiuk is this a point to contribute? I want to submit a PR for it.

I believe this is far more complex than that - looks like auto-balancing of work done by triggerer. And it requires an AIP (airflow improvement proposal) to discuss various cases and consequences. This will likely be non-trivial change. But I am not sure - as I am not really expert not have not thought about it too much . We are however discussing now a proposal of making deferrable "default" in Airflow 3 and this might become more important topic - so I invite you to chime in and ask your question in https://lists.apache.org/thread/90755fvgr1912kgj2h47og2rof55dc5t to show that discussing, solving cases like this are also important to think of deferrable workflow are "really ready" and community being ready to answer questions like that.