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

#
#
from __future__ import annotations
import asyncio
from functools import partial
from typing import Any, Sequence
from asgiref.sync import sync_to_async
from airflow import AirflowException
from airflow.providers.apache.kafka.hooks.consume import KafkaConsumerHook
from airflow.triggers.base import BaseTrigger, TriggerEvent
from airflow.utils.module_loading import import_string
class AwaitMessageTrigger(BaseTrigger):