You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "zoid-w (via GitHub)" <gi...@apache.org> on 2023/06/08 14:24:35 UTC

[GitHub] [airflow] zoid-w opened a new issue, #31795: AwaitMessageTriggerFunctionSensor not firing all eligble messages

zoid-w opened a new issue, #31795:
URL: https://github.com/apache/airflow/issues/31795

   ### Apache Airflow version
   
   2.6.1
   
   ### What happened
   
   The AwaitMessageTriggerFunctionSensor is showing some buggy behaviour.
   When consuming from a topic, it is correctly applying the apply_function in order to yield a TriggerEvent.
   
   However, it is consuming multiple messages at a time and not yielding a trigger for the correct amount of messages that would be eligble (return a value in the apply_function). The observed behaviour is as follows: 
   - Sensor is deferred and messages start getting consumed
   - Multiple eligble messages trigger a single TriggerEvent instead of multiple TriggerEvents.
   - The sensor returns to a deferred state , repeating the cycle.
   
   The event_triggered_function is being called correctly. However, due to the issue in consuming and correctly generating the appropriate TriggerEvents some of them are missed.
   
   ### What you think should happen instead
   
   Each eligble message should create an individual TriggerEvent to be consumed by the event_triggered_function.
   
   ### How to reproduce
   
   - Use a producer DAG to produce a set amount of messages on your kafka topic
   - Use a listener DAG to consume this topic, screening for eligble messages (apply_function) and use the event_trigger_function to monitor the amount of events that are being consumed.
   
   ### Operating System
   
   Kubernetes cluster - Linux
   
   ### Versions of Apache Airflow Providers
   
   apache-airflow-providers-apache-kafka==1.1.0
   
   ### Deployment
   
   Official Apache Airflow Helm Chart
   
   ### Deployment details
   
   helm chart version 1.9.0
   
   ### Anything else
   
   Every time (independent of topic, message content, apply_function and event_triggered_function)
   
   ### Are you willing to submit PR?
   
   - [ ] Yes I am willing to submit a PR!
   
   ### Code of Conduct
   
   - [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
   


-- 
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: commits-unsubscribe@airflow.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [airflow] ikholodkov commented on issue #31795: AwaitMessageTriggerFunctionSensor not firing all eligble messages

Posted by "ikholodkov (via GitHub)" <gi...@apache.org>.
ikholodkov commented on issue #31795:
URL: https://github.com/apache/airflow/issues/31795#issuecomment-1612147940

   I have the same issue. I'm using this for triggering another DAG. Tried with different amounts of DagRuns and every time I lost messages. Looks like some messages was committed by consumer but not processed by apply_function.  
   Also, sometimes I got this error
   `cimpl.KafkaException: KafkaError{code=_NO_OFFSET,val=-168,str="Commit failed: Local: No offset stored"}`


-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [airflow] hussein-awala commented on issue #31795: AwaitMessageTriggerFunctionSensor not firing all eligble messages

Posted by "hussein-awala (via GitHub)" <gi...@apache.org>.
hussein-awala commented on issue #31795:
URL: https://github.com/apache/airflow/issues/31795#issuecomment-1627712416

   @zoid-w @ikholodkov Do you have multiple triggerers running simultaneously with the `AIRFLOW__SCHEDULER__JOB_HEARTBEAT_SEC` value set to 30 or higher?
   
   Regarding #31803, it addresses an issue when there is a single triggerer under pressure. Previously, it could take some time to process sent events, allowing the triggerer to read another message in the meantime. However, after this PR, that scenario is no longer possible.
   
   It's important to note that the triggerer does not provide an exactly-once guarantee for running triggers; its guarantee is at least-once. This means it is possible to have the same trigger running on multiple triggerers simultaneously. We encountered an issue with the HA mode, which I have fixed in #32123. So, when version 2.6.3 is released, it is advisable to upgrade to it.
   
   While #31803 and #32123 can significantly reduce the probability of this problem occurring to less than 1%, it is still a possibility. I am currently exploring ways to synchronize between the triggerers when necessary.


-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [airflow] potiuk commented on issue #31795: AwaitMessageTriggerFunctionSensor not firing all eligble messages

Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on issue #31795:
URL: https://github.com/apache/airflow/issues/31795#issuecomment-1619218979

   Sure. Watch the devlist. We release providers roughly every 2 weeks. You can watch the devlist for announcements.


-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [airflow] potiuk closed issue #31795: AwaitMessageTriggerFunctionSensor not firing all eligble messages

Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk closed issue #31795: AwaitMessageTriggerFunctionSensor not firing all eligble messages
URL: https://github.com/apache/airflow/issues/31795


-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [airflow] ikholodkov commented on issue #31795: AwaitMessageTriggerFunctionSensor not firing all eligble messages

Posted by "ikholodkov (via GitHub)" <gi...@apache.org>.
ikholodkov commented on issue #31795:
URL: https://github.com/apache/airflow/issues/31795#issuecomment-1614896859

   Sorry for writing in closed issue. Could someone tell how could I know when linked [bugfix](https://github.com/apache/airflow/pull/31803) will be released (via new Airflow or new Kafka provider versions)?


-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [airflow] hussein-awala commented on issue #31795: AwaitMessageTriggerFunctionSensor not firing all eligble messages

Posted by "hussein-awala (via GitHub)" <gi...@apache.org>.
hussein-awala commented on issue #31795:
URL: https://github.com/apache/airflow/issues/31795#issuecomment-1583215431

   Since you already have the infrastructure in place to experiment with this sensor, could you please test whether #31803 resolves this issue?


-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [airflow] zoid-w commented on issue #31795: AwaitMessageTriggerFunctionSensor not firing all eligble messages

Posted by "zoid-w (via GitHub)" <gi...@apache.org>.
zoid-w commented on issue #31795:
URL: https://github.com/apache/airflow/issues/31795#issuecomment-1592964558

   Hi, I have done some testing with #31803. We can see the following behavior.
   I used a test setup that is expected to generate 25 events from 40 messages (so only 25/40 messages are eligible ones).
   ## Logs : 
   ```
   [2023-06-15, 13:56:37 CEST] {triggerer_job_runner.py:608} INFO - Trigger ***/manual__2023-06-15T11:56:14.094686+00:00/listen_for_message/-1/1 (ID 129) fired: TriggerEvent<'FOO'>
   [2023-06-15, 13:56:37 CEST] {triggerer_job_runner.py:608} INFO - Trigger ***/manual__2023-06-15T11:56:14.094686+00:00/listen_for_message/-1/1 (ID 129) fired: TriggerEvent<'FOO'>
   [2023-06-15, 13:57:23 CEST] {triggerer_job_runner.py:608} INFO - Trigger ***/manual__2023-06-15T11:56:14.094686+00:00/listen_for_message/-1/1 (ID 130) fired: TriggerEvent<'FOO'>
   [2023-06-15, 13:58:11 CEST] {triggerer_job_runner.py:608} INFO - Trigger ***/manual__2023-06-15T11:56:14.094686+00:00/listen_for_message/-1/1 (ID 131) fired: TriggerEvent<'FOO'>
   [2023-06-15, 13:58:56 CEST] {triggerer_job_runner.py:608} INFO - Trigger ***/manual__2023-06-15T11:56:14.094686+00:00/listen_for_message/-1/1 (ID 132) fired: TriggerEvent<'BAR'>
   [2023-06-15, 13:58:56 CEST] {triggerer_job_runner.py:608} INFO - Trigger ***/manual__2023-06-15T11:56:14.094686+00:00/listen_for_message/-1/1 (ID 132) fired: TriggerEvent<'FOO'>
   [2023-06-15, 13:58:56 CEST] {triggerer_job_runner.py:608} INFO - Trigger ***/manual__2023-06-15T11:56:14.094686+00:00/listen_for_message/-1/1 (ID 132) fired: TriggerEvent<'FOO'>
   [2023-06-15, 13:59:41 CEST] {triggerer_job_runner.py:608} INFO - Trigger ***/manual__2023-06-15T11:56:14.094686+00:00/listen_for_message/-1/1 (ID 133) fired: TriggerEvent<'FOO'>
   [2023-06-15, 13:59:41 CEST] {triggerer_job_runner.py:608} INFO - Trigger ***/manual__2023-06-15T11:56:14.094686+00:00/listen_for_message/-1/1 (ID 133) fired: TriggerEvent<'BAR'>
   [2023-06-15, 14:00:31 CEST] {triggerer_job_runner.py:608} INFO - Trigger ***/manual__2023-06-15T11:56:14.094686+00:00/listen_for_message/-1/1 (ID 134) fired: TriggerEvent<'BAR'>
   [2023-06-15, 14:00:31 CEST] {triggerer_job_runner.py:608} INFO - Trigger ***/manual__2023-06-15T11:56:14.094686+00:00/listen_for_message/-1/1 (ID 134) fired: TriggerEvent<'FOO'>
   [2023-06-15, 14:00:31 CEST] {triggerer_job_runner.py:608} INFO - Trigger ***/manual__2023-06-15T11:56:14.094686+00:00/listen_for_message/-1/1 (ID 134) fired: TriggerEvent<'BAR'>
   [2023-06-15, 14:00:31 CEST] {triggerer_job_runner.py:608} INFO - Trigger ***/manual__2023-06-15T11:56:14.094686+00:00/listen_for_message/-1/1 (ID 134) fired: TriggerEvent<'FOO'>
   [2023-06-15, 14:00:31 CEST] {triggerer_job_runner.py:608} INFO - Trigger ***/manual__2023-06-15T11:56:14.094686+00:00/listen_for_message/-1/1 (ID 134) fired: TriggerEvent<'FOO'>
   [2023-06-15, 14:00:31 CEST] {triggerer_job_runner.py:608} INFO - Trigger ***/manual__2023-06-15T11:56:14.094686+00:00/listen_for_message/-1/1 (ID 134) fired: TriggerEvent<'BAR'>
   [2023-06-15, 14:00:31 CEST] {triggerer_job_runner.py:608} INFO - Trigger ***/manual__2023-06-15T11:56:14.094686+00:00/listen_for_message/-1/1 (ID 134) fired: TriggerEvent<'FOO'>
   [2023-06-15, 14:01:22 CEST] {triggerer_job_runner.py:608} INFO - Trigger ***/manual__2023-06-15T11:56:14.094686+00:00/listen_for_message/-1/1 (ID 135) fired: TriggerEvent<'FOO'>
   [2023-06-15, 14:02:13 CEST] {triggerer_job_runner.py:608} INFO - Trigger ***/manual__2023-06-15T11:56:14.094686+00:00/listen_for_message/-1/1 (ID 136) fired: TriggerEvent<'FOO'>
   [2023-06-15, 14:02:13 CEST] {triggerer_job_runner.py:608} INFO - Trigger ***/manual__2023-06-15T11:56:14.094686+00:00/listen_for_message/-1/1 (ID 136) fired: TriggerEvent<'BAR'>
   [2023-06-15, 14:02:13 CEST] {triggerer_job_runner.py:608} INFO - Trigger ***/manual__2023-06-15T11:56:14.094686+00:00/listen_for_message/-1/1 (ID 136) fired: TriggerEvent<'FOO'>
   [2023-06-15, 14:02:13 CEST] {triggerer_job_runner.py:608} INFO - Trigger ***/manual__2023-06-15T11:56:14.094686+00:00/listen_for_message/-1/1 (ID 136) fired: TriggerEvent<'BAR'>
   [2023-06-15, 14:02:13 CEST] {triggerer_job_runner.py:608} INFO - Trigger ***/manual__2023-06-15T11:56:14.094686+00:00/listen_for_message/-1/1 (ID 136) fired: TriggerEvent<'FOO'>
   [2023-06-15, 14:02:13 CEST] {triggerer_job_runner.py:608} INFO - Trigger ***/manual__2023-06-15T11:56:14.094686+00:00/listen_for_message/-1/1 (ID 136) fired: TriggerEvent<'BAR'>
   [2023-06-15, 14:02:59 CEST] {triggerer_job_runner.py:608} INFO - Trigger ***/manual__2023-06-15T11:56:14.094686+00:00/listen_for_message/-1/1 (ID 137) fired: TriggerEvent<'BAR'>
   [2023-06-15, 14:02:59 CEST] {triggerer_job_runner.py:608} INFO - Trigger ***/manual__2023-06-15T11:56:14.094686+00:00/listen_for_message/-1/1 (ID 137) fired: TriggerEvent<'FOO'>
   ```
   
   What we can see is that only 9 TriggerEvents are created (based upon ID). In this case you event_triggered_function will therefore only be called 9 times instead of 25. From the timestamp I suspect this issue is occuring since the TriggerEvents are generated at exactly the same time, yielding mutiple TriggerEvents with the same ID since the TriggerEvent ID has not yet been increased.
   
   From the discussion in #31703 I am unsure what the correct approach here would be.
   


-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [airflow] potiuk commented on issue #31795: AwaitMessageTriggerFunctionSensor not firing all eligble messages

Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on issue #31795:
URL: https://github.com/apache/airflow/issues/31795#issuecomment-1585629238

   cc: @dylanbstorey -> FYI maybe you can also help :D


-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [airflow] potiuk commented on issue #31795: AwaitMessageTriggerFunctionSensor not firing all eligble messages

Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on issue #31795:
URL: https://github.com/apache/airflow/issues/31795#issuecomment-1685356170

   Looked at linked PR that closed it https://github.com/apache/airflow/pull/31803 and see in Kafka provider's release notes https://airflow.apache.org/docs/apache-airflow-providers-apache-kafka/stable/changelog.html which Kafka providerhas the fix - then - install that provider.  You can also try 2.7.0 if you do not want to upgrade selectively just Kafka provider - it's been released on Friday and when you use constraints, it should use the last version of Kafka provider released then (look at constraint files and you will see which version is linked by "default" to wich version of Airflow.
   
   See https://airflow.apache.org/docs/apache-airflow/stable/installation/installing-from-pypi.html#installation-and-upgrade-scenarios for all scenarios where you can upgrade airflow and providers independently.
    


-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [airflow] pauloventurab commented on issue #31795: AwaitMessageTriggerFunctionSensor not firing all eligble messages

Posted by "pauloventurab (via GitHub)" <gi...@apache.org>.
pauloventurab commented on issue #31795:
URL: https://github.com/apache/airflow/issues/31795#issuecomment-1684896249

   Hello guys, sorry to write in this closed topic, but I'm not sure what is the conclusion. 
   Fixed? I still have problems on 2.6.3
   Not fixed? What is the action plan ? Do you need some help?
   
   Thanks and have a good life!


-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [airflow] zoid-w commented on issue #31795: AwaitMessageTriggerFunctionSensor not firing all eligble messages

Posted by "zoid-w (via GitHub)" <gi...@apache.org>.
zoid-w commented on issue #31795:
URL: https://github.com/apache/airflow/issues/31795#issuecomment-1589078136

   Will be doing some testing checking whether the commits fix this issue at the end of the week!


-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [airflow] boring-cyborg[bot] commented on issue #31795: AwaitMessageTriggerFunctionSensor not firing all eligble messages

Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on issue #31795:
URL: https://github.com/apache/airflow/issues/31795#issuecomment-1582680030

   Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.
   


-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [airflow] ikholodkov commented on issue #31795: AwaitMessageTriggerFunctionSensor not firing all eligble messages

Posted by "ikholodkov (via GitHub)" <gi...@apache.org>.
ikholodkov commented on issue #31795:
URL: https://github.com/apache/airflow/issues/31795#issuecomment-1634317356

   @hussein-awala are you talking about the situation when a sensor could commit message twice? I mean, is it linked to [confluent-kafka-python](https://github.com/confluentinc/confluent-kafka-python) error i've mentioned higher? 
   `cimpl.KafkaException: KafkaError{code=_NO_OFFSET,val=-168,str="Commit failed: Local: No offset stored"}` 
   
   Am asking because in Airflow 2.6.3 and `apache-airflow-providers-apache-kafka==1.1.2` i've faced this problem again and ready to open an issue. 


-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [airflow] zoid-w commented on issue #31795: AwaitMessageTriggerFunctionSensor not firing all eligble messages

Posted by "zoid-w (via GitHub)" <gi...@apache.org>.
zoid-w commented on issue #31795:
URL: https://github.com/apache/airflow/issues/31795#issuecomment-1613348472

   The apply_function is only called on a TriggerEvent that get's yielded from the AwaitMessageTrigger. The issue to me seems to be that we are yielding multiple TriggerEvents (with the same ID). I assume Airflow is then de-duplicating them (which is as stated in the [docs](https://airflow.apache.org/docs/apache-airflow/stable/authoring-and-scheduling/deferring.html#high-availability) to be excepted). However, if the de-duplication is using the TriggerEvent __eq__ method it should not de-duplicate them (as the payload is part of the comparison, which in my case contains the message_offset which should be unique). Not sure how the de-duplication is done though.
   


-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [airflow] hussein-awala commented on issue #31795: AwaitMessageTriggerFunctionSensor not firing all eligble messages

Posted by "hussein-awala (via GitHub)" <gi...@apache.org>.
hussein-awala commented on issue #31795:
URL: https://github.com/apache/airflow/issues/31795#issuecomment-1595846695

   @zoid-w This is very strange, where the `break` statement after yielding the event should be sufficient to ensure that the trigger will not process any other messages and therefore will not yield a new event.
   
   I will run some tests to check your assumption.


-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org