Skip to content

AwaitMessageTriggerFunctionSensor not firing all eligble messages #31795

New issue
New issue
Closed
#31803
Closed
AwaitMessageTriggerFunctionSensor not firing all eligble messages#31795
#31803
Labels
area:providerskind:bugThis is a clearly a bug

Description

@zoid-w
zoid-w
opened on Jun 8, 2023

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

Activity

  1. zoid-w
    added
    area:core
    kind:bugThis is a clearly a bug
    needs-triagelabel for new issues that we didn't triage yet
    on Jun 8, 2023

  2. boring-cyborg commented on Jun 8, 2023

    @boring-cyborg
    boring-cyborgBot
    on Jun 8, 2023

    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.

  3. hussein-awala
    added
    area:providers
    provider:Apache
    and removed
    area:core
    needs-triagelabel for new issues that we didn't triage yet
    on Jun 8, 2023

  4. hussein-awala commented on Jun 8, 2023

    @hussein-awala
    hussein-awala
    on Jun 8, 2023
    Member

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

  5. potiuk commented on Jun 10, 2023

    @potiuk
    potiuk
    on Jun 10, 2023
    Member

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

  6. zoid-w commented on Jun 13, 2023

    @zoid-w
    zoid-w
    on Jun 13, 2023
    Author

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

  7. zoid-w commented on Jun 15, 2023

    @zoid-w
    zoid-w
    on Jun 15, 2023
    Author

    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.

  8. hussein-awala commented on Jun 17, 2023

    @hussein-awala
    hussein-awala
    on Jun 17, 2023
    Member

    @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.

  9. ikholodkov commented on Jun 28, 2023

    @ikholodkov
    ikholodkov
    on Jun 28, 2023

    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"}

  10. zoid-w commented on Jun 29, 2023

    @zoid-w
    zoid-w
    on Jun 29, 2023
    Author

    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 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.

  11. potiuk
    closed this as completedin #31803on Jun 30, 2023

  12. ikholodkov commented on Jun 30, 2023

    @ikholodkov
    ikholodkov
    on Jun 30, 2023

    Sorry for writing in closed issue. Could someone tell how could I know when linked bugfix will be released (via new Airflow or new Kafka provider versions)? I'm talking about release plans.

  13. potiuk commented on Jul 3, 2023

    @potiuk
    potiuk
    on Jul 3, 2023
    Member

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

  14. hussein-awala commented on Jul 9, 2023

    @hussein-awala
    hussein-awala
    on Jul 9, 2023
    Member

    @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.

  15. ikholodkov commented on Jul 13, 2023

    @ikholodkov
    ikholodkov
    on Jul 13, 2023

    @hussein-awala are you talking about the situation when a sensor could commit message twice? I mean, is it linked to confluent-kafka-python error i've mentioned above?
    cimpl.KafkaException: KafkaError{code=_NO_OFFSET,val=-168,str="Commit failed: Local: No offset stored"}

    I'm asking because in Airflow 2.6.3 and apache-airflow-providers-apache-kafka==1.1.2 i've faced this again and opened the #32585.

  16. pauloventurab commented on Aug 19, 2023

    @pauloventurab
    pauloventurab
    on Aug 19, 2023

    Hello guys, sorry to write in this closed topic, but I'm not sure what is the conclusion.

    I have a consumer, that expect hundreds of messages in seconds, I want to send each one of those messages to a ML workflow in another dag. The issue now is that not all messages is being send, just the first one.

    Fixed? I still have problems on 2.6.3
    Not fixed? What is the action plan ? Do you need some help?

    If not fixed, have we some idea how could I do it? even with expand, but I'm facing some issues on it.

    Thanks and have a good life!

  17. potiuk commented on Aug 20, 2023

    @potiuk
    potiuk
    on Aug 20, 2023
    Member

    Looked at linked PR that closed it #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.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Labels

    area:providerskind:bugThis is a clearly a bug

    Type

    No type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions