Trigger runner process locked with multiple Workflow triggers

[ArvidMartensRenson]

Apache Airflow version

3.0.0

If "Other Airflow 2 version" selected, which one?

No response

What happened?

I am running a few dags that can start multiple workflow triggers. When the amount of triggers is low, below 10 or so; all works fine but whenever it goes higher the trigger runner seems to get stuck. I traced it back to one of the dag_count calls on RuntimeTaskInstance getting stuck/taking too long which locks all the other triggers and the syncing of the trigger runner as the SUPERVISOR_COMMS object stays in its locked state. The triggers seems to be started at the same time.

I already tried to solve the problem by introducing a timeout on the count calls but that made the trigger runner crash as the incoming messages are getting mixed up (response of a cancelled call finally getting through). Only solution I see right now is to perform API calls on the public endpoint for the dag count method calls.

What you think should happen instead?

No response

How to reproduce

Have more than twenty or so workflow triggers running started at the same, I suppose they poll states at about the same time.

Operating System

debian 12

Versions of Apache Airflow Providers

No response

Deployment

Official Apache Airflow Helm Chart

Deployment details

No response

Anything else?

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[uranusjr]

Is it possible to write a dag that can reliably reproduce this?

[kaxil]

cc @gopidesupavan

[gopidesupavan]

Yeah will look at this today.

[vatsrahul1001]

@gopidesupavan are you still looking into this?

[ArvidMartensRenson]

For now I solved througt the creation of an async Task SDK client which the triggers use directly. In my setup the triggers always have the api server running on the same instances, hence I could use localhost traffic. In this way it is rather similar like the in-process traffic but I do not overload the data pipe between trigger runner and supervisor. From my investigation it seems that using too much external workload triggers (count calls) together with the supervisor-runner syncs congests that pipe in such a way that all calls were blocked indefinitely resulting in a trigger process that seems alive (heartbeats still worked) but in reality it is stuck.

Probably this is not the designed pattern, but for now this solution makes it possible for me to role out airflow 3.0 at our office.

[gopidesupavan]

I am able to re produce it with one of example ExternalTaskSensor dags. It looks blocking somewhere and triggers are not getting to added to further process.

Need to workout why it's blocking though we use async lock. Will check and update.

[gopidesupavan]

For now I solved througt the creation of an async Task SDK client which the triggers use directly. In my setup the triggers always have the api server running on the same instances, hence I could use localhost traffic. In this way it is rather similar like the in-process traffic but I do not overload the data pipe between trigger runner and supervisor. From my investigation it seems that using too much external workload triggers (count calls) together with the supervisor-runner syncs congests that pipe in such a way that all calls were blocked indefinitely resulting in a trigger process that seems alive (heartbeats still worked) but in reality it is stuck.

Probably this is not the designed pattern, but for now this solution makes it possible for me to role out airflow 3.0 at our office.

@ArvidMartensRenson we dont have async client in Task SDK, did you overwrite the Triggerer client method, does it fixed?

[ArvidMartensRenson]

I created an async client from the sync client, using httpx.AsyncClient as the base (like httpx.Client is for the sync client). Inherited the init client from the sync task client. Only problem I had was that the operations had to be rewritten as an async method, so I copied the code from the sync client and changed the method definition to async. I found this implementation nicer than using plain api calls with the async httpx client.

Traffic is over localhost so even then a token is required, I reused the methods that are used in the creation of the task workload to get the token. I considered dropping the token requirement but it seemed safer and whenever the trigger service and api server are separated it will be needed.

I implemented this solution last Thursday and for now the trigger service has not been stuck and the external workload triggers all completed. So for now I would conclude that the solution works.

If it is desirable to have an async client I am happy to help on creating the client, however I am a little worried with duplicate code I have now, same code for the sync and async client. If someone knows a solution for this or a way to handle that problem, that would be nice.

For the record, I am using a custom implementation of the workload trigger. I was an early adopter of the trigger paradigm and the external workload trigger did not exist at that time, so we had to implement our own trigger. We never came around converting to the standard provider implementation.