From 8fc0407498b7167244501607f1003b294c694858 Mon Sep 17 00:00:00 2001
From: Mahendra M <mahendra.m@gmail.com>
Date: Tue, 28 May 2013 11:51:57 +0530
Subject: [PATCH] Auto commit timer is not periodic

The auto commit timer is one-shot. After the first commit, it does
not fire again. This ticket fixes the issue.

Also, in util.ReentrantTimer(), some duplicate code was cleaned up
---
 kafka/consumer.py | 25 ++++++++++++++++++-------
 kafka/util.py     | 11 +++++------
 2 files changed, 23 insertions(+), 13 deletions(-)

diff --git a/kafka/consumer.py b/kafka/consumer.py
index f1231133c..5c39cb783 100644
--- a/kafka/consumer.py
+++ b/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
diff --git a/kafka/util.py b/kafka/util.py
index 5dc6bc21b..8c02cb29a 100644
--- a/kafka/util.py
+++ b/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