feat: commit offsets on successfull read (#11)

gorgias · Jan 9, 2025 · 35836b1 · 35836b1
1 parent 401c25b
commit 35836b1
Showing 1 changed file with 4 additions and 0 deletions.
diff --git a/bizon/connectors/sources/kafka/src/source.py b/bizon/connectors/sources/kafka/src/source.py
@@ -360,6 +360,10 @@ def read_topic(self, pagination: dict = None) -> SourceIteration:
                 records=[],
             )
 
+        # Commit offsets so we keep track of conumer-group progress in Confluent Cloud
+        # It also allows us to leverage Datadog lag monitors
+        self.consumer.commit(asynchronous=False)
+
         return SourceIteration(
             next_pagination=self.topic_offsets.model_dump(),
             records=records,