(airflow) branch main updated: Provide the consumed message to consumer.commit in AwaitMessageTrigger (#36272)

[husseinawala]

This is an automated email from the ASF dual-hosted git repository.

husseinawala pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git


The following commit(s) were added to refs/heads/main by this push:
     new 148233a19e Provide the consumed message to consumer.commit in 
AwaitMessageTrigger (#36272)
148233a19e is described below

commit 148233a19ea68f424a7077d3bba6e6ca81679c10
Author: Hussein Awala <huss...@awala.fr>
AuthorDate: Mon Dec 18 11:22:49 2023 +0100

    Provide the consumed message to consumer.commit in AwaitMessageTrigger 
(#36272)
---
 airflow/providers/apache/kafka/triggers/await_message.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/airflow/providers/apache/kafka/triggers/await_message.py 
b/airflow/providers/apache/kafka/triggers/await_message.py
index c13a3a59ca..0d2d69300b 100644
--- a/airflow/providers/apache/kafka/triggers/await_message.py
+++ b/airflow/providers/apache/kafka/triggers/await_message.py
@@ -108,9 +108,9 @@ class AwaitMessageTrigger(BaseTrigger):
             else:
                 rv = await async_message_process(message)
                 if rv:
-                    await async_commit(asynchronous=False)
+                    await async_commit(message=message, asynchronous=False)
                     yield TriggerEvent(rv)
                     break
                 else:
-                    await async_commit(asynchronous=False)
+                    await async_commit(message=message, asynchronous=False)
                     await asyncio.sleep(self.poll_interval)