fix: Do not run asynchronous commit in thread pool executor.

airtai · Dec 21, 2024 · e1eb153 · e1eb153
1 parent 9caefa7
commit e1eb153
Showing 1 changed file with 5 additions and 1 deletion.
diff --git a/faststream/confluent/client.py b/faststream/confluent/client.py
@@ -350,7 +350,11 @@ async def start(self) -> None:
 
     async def commit(self, asynchronous: bool = True) -> None:
         """Commits the offsets of all messages returned by the last poll operation."""
-        await call_or_await(self.consumer.commit, asynchronous=asynchronous)
+        if asynchronous:
+            # Asynchronous commit is non-blocking:
+            self.consumer.commit(asynchronous=True)
+        else:
+            await call_or_await(self.consumer.commit, asynchronous=False)
 
     async def stop(self) -> None:
         """Stops the Kafka consumer and releases all resources."""