Break AwaitMessageTrigger execution when finding a message with the d…

…esired format (#31803)

Signed-off-by: Hussein Awala <hussein@awala.fr>
Co-authored-by: eladkal <45845474+eladkal@users.noreply.github.com>

airflow/providers/apache/kafka/triggers/await_message.py

@@ -113,6 +113,7 @@ async def run(self):
                 if rv:
                     await async_commit(asynchronous=False)
                     yield TriggerEvent(rv)
+                    break
                 else:
                     await async_commit(asynchronous=False)
                     await asyncio.sleep(self.poll_interval)