Move fetching last known offset logic to a stand alone function.

zackdever · zackdever · commit eecea8809ec7 · 2014-08-25T12:00:19.000-07:00
The `Consumer` class fetches the last known offsets in `__init__` if `auto_commit` is enabled, but it would be nice to expose this behavior for consumers that aren't using auto_commit. This doesn't change existing behavior, just exposes the ability to easily fetch and set the last known offsets. Once dpkp#162 or something similar lands this may no longer be necessary, but it looks like that might take a while to make it through.
diff --git a/kafka/consumer.py b/kafka/consumer.py
@@ -100,23 +100,29 @@ def __init__(self, client, group, topic, partitions=None, auto_commit=True,
                                                self.commit)
             self.commit_timer.start()
 
+        if auto_commit:
+            self.fetch_last_known_offsets(partitions)
+        else:
+            for partition in partitions:
+                self.offsets[partition] = 0
+
+    def fetch_last_known_offsets(self, partitions=None):
+        if not partitions:
+            partitions = self.client.topic_partitions[self.topic]
+
         def get_or_init_offset_callback(resp):
             try:
                 kafka.common.check_error(resp)
                 return resp.offset
             except kafka.common.UnknownTopicOrPartitionError:
                 return 0
 
-        if auto_commit:
-            for partition in partitions:
-                req = OffsetFetchRequest(topic, partition)
-                (offset,) = self.client.send_offset_fetch_request(group, [req],
-                              callback=get_or_init_offset_callback,
-                              fail_on_error=False)
-                self.offsets[partition] = offset
-        else:
-            for partition in partitions:
-                self.offsets[partition] = 0
+        for partition in partitions:
+            req = OffsetFetchRequest(self.topic, partition)
+            (offset,) = self.client.send_offset_fetch_request(self.group, [req],
+                          callback=get_or_init_offset_callback,
+                          fail_on_error=False)
+            self.offsets[partition] = offset
 
     def commit(self, partitions=None):
         """
