Auto commit timer is not periodic

kafka/consumer.py

@@ -34,18 +34,20 @@ def __init__(self, client, group, topic, auto_commit=False, auto_commit_every_n=
         self.client._load_metadata_for_topics(topic)
         self.offsets = {}
 
-        # Set up the auto-commit timer
-        if auto_commit is True:
-            if auto_commit_every_t is not None:
-                self.commit_timer = ReentrantTimer(auto_commit_every_t, self.commit)
-                self.commit_timer.start()
-
+        # Variables for handling offset commits
         self.commit_lock = Lock()
+        self.commit_timer = None
         self.count_since_commit = 0
         self.auto_commit = auto_commit
         self.auto_commit_every_n = auto_commit_every_n
         self.auto_commit_every_t = auto_commit_every_t
-
+
+        # Set up the auto-commit timer
+        if auto_commit is True and auto_commit_every_t is not None:
+            self.commit_timer = ReentrantTimer(auto_commit_every_t,
+                                               self._timed_commit)
+            self.commit_timer.start()
+
         def get_or_init_offset_callback(resp):
             if resp.error == ErrorMapping.NO_ERROR:
                 return resp.offset
@@ -101,6 +103,15 @@ def seek(self, offset, whence):
         else:
             raise ValueError("Unexpected value for `whence`, %d" % whence)
 
+    def _timed_commit(self):
+        """
+        Commit offsets as part of timer
+        """
+        self.commit()
+
+        # Once the commit is done, start the timer again
+        self.commit_timer.start()
+
     def commit(self, partitions=[]):
         """
         Commit offsets for this consumer

kafka/util.py

@@ -71,13 +71,12 @@ def __init__(self, t, fn):
         self.fn = fn
 
     def start(self):
-        if self.timer is None:
-            self.timer = Timer(self.t / 1000., self.fn)
-            self.timer.start()
-        else:
+        if self.timer is not None:
             self.timer.cancel()
-            self.timer = Timer(self.t / 1000., self.fn)
-            self.timer.start()
+
+        self.timer = Timer(self.t / 1000., self.fn)
+        self.timer.start()
 
     def stop(self):
         self.timer.cancel()
+        self.timer = None