Provide the consumed message to consumer.commit in AwaitMessageTrigger (

apache#36272)
romsharon98 · Dec 18, 2023 · 148233a · 148233a
1 parent 41096e0
commit 148233a
Showing 1 changed file with 2 additions and 2 deletions.
diff --git a/airflow/providers/apache/kafka/triggers/await_message.py b/airflow/providers/apache/kafka/triggers/await_message.py
@@ -108,9 +108,9 @@ async def run(self):
             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)