AwaitMessageTriggerFunctionSensor take long time to trigger event_triggered_function

[hungnguyenthien]

Apache Airflow version

2.7.1

What happened

Hi guys,

i'm using the AwaitMessageTriggerFunctionSensor with the following example on your Docs: https://github.com/apache/airflow/blob/providers-apache-kafka/1.1.2/tests/system/providers/apache/kafka/example_dag_event_listener.py

The issue is when I trigger 'fizzbuzz-load-topic' run only take 5 seconds to load kafka connection config and push 50 messages to the topic 'fizz_buzz' but the 'fizz-buzz' dag take up to 10 minutes to be triggered 3 times by 'pick_downstream_dag' task. The amount of time from when the last message was sent until it was processed by 'fizzbuzz-listener-dag' is very long.

Can you give a hand here please?

Thanks

What you think should happen instead

I expected the dag 'fizz-buzz' will be triggered near real time, I try to modify poll_timeout and poll_interval to 0.00001 and 0.00005 second to be able to achieve that but I can't. Did I miss anything?

How to reproduce

I don't know how to help improve it.

Operating System

Python 3.8 Ubuntu 20.04 - Linux

Versions of Apache Airflow Providers

apache-airflow-providers-apache-kafka==1.1.2

Deployment

apache-airflow[postgres, rabbitmq, celery, crypto]==2.7.1

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

[Taragolis]

dag take up to 10 minutes to be triggered 3 times by 'pick_downstream_dag' task.

https://airflow.apache.org/docs/apache-airflow/stable/index.html#why-not-airflow

---

I have a look at the example, and either it is a bad example and should not use

airflow/tests/system/providers/apache/kafka/example_dag_event_listener.py

Lines 110 to 113 in ba0bab0

	print(f"encountered {message} - executing external dag!")
	TriggerDagRunOperator(trigger_dag_id="fizz-buzz", task_id=f"{message}{_generate_uuid()}").execute(
	context
	)

🤦 🤦 🤦

[hungnguyenthien]

@Taragolis thanks for help. But there are a few things I want to clarify:

1. When I see log of 'fizzbuzz-listener-dag' on UI, I can see the message log "encountered X FIZZ/BUZZ/executing external dag!" emitted from 'pick_downstream_dag' function approximately every 45 seconds, and it looks like the process slept for 45 seconds without doing anything. Please see the log below to see that:

It would be great if I could shorten this time from 45s to 10ms. Please help me for that.

2. Is there another way besides using AwaitMessageTriggerFunctionSensor to achieve the same thing with Airflow but faster?

[Taragolis]

emitted from 'pick_downstream_dag' function approximately every 45 seconds, and it looks like the process slept for 45 seconds without doing anything. Please see the log below to see that:

Unfortunetly this example contain worst advice in Airflow, somehow this was missed during PR when Kafka provider was added. Sometimes it happen we all humans and could miss something. But I hope this sample would be removed in the next release of provider, see: #34470

It would be great if I could shorten this time from 45s to 10ms. Please help me for that.

10ms it is not possible due to initial design and purpose of Airflow: run Batch tasks, something might (or might not) change in the future but right now we have what we have.
Even if rewrite this part to to use Rest API then it would be more close to the seconds, rather then milliseconds

[hungnguyenthien]

Thank you very much for the answers. I'm clear !