### 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)

"""A trigger that waits for a message matching specific criteria to arrive in Kafka.
    The behavior of the consumer of this trigger is as follows:
    - poll the Kafka topics for a message, if no message returned, sleep
    - process the message with provided callable and commit the message offset:
        - if callable returns any data, raise a TriggerEvent with the return data
        - else continue to next message
    :param kafka_config_id: The connection object to use, defaults to "kafka_default"
    :param topics: The topic (or topic regex) that should be searched for messages
    :param apply_function: the location of the function to apply to messages for determination of matching
        criteria. (In python dot notation as a string)
    :param apply_function_args: A set of arguments to apply to the callable, defaults to None
    :param apply_function_kwargs: A set of key word arguments to apply to the callable, defaults to None,
        defaults to None
    :param poll_timeout: How long the Kafka client should wait before returning from a poll request to
        Kafka (seconds), defaults to 1
    :param poll_interval: How long the the trigger should sleep after reaching the end of the Kafka log
        (seconds), defaults to 5
    """
    def __init__(