Spawn the commit thread only if necessary

If there are no messages being consumed, the timer keeps
creating new threads at the specified intervals. This may
not be necessary. We can control this behaviour such that
the timer thread is started only when a message is consumed

Author: mahendra

kafka/consumer.py

@@ -59,7 +59,6 @@ def __init__(self, client, group, topic, auto_commit=True,
         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:
@@ -157,7 +156,7 @@ def _timed_commit(self):
         self.commit()
 
         # Once the commit is done, start the timer again
-        self.commit_timer.start()
+        # self.commit_timer.start()
 
     def commit(self, partitions=[]):
         """
@@ -224,6 +223,10 @@ def __iter__(self):
             if len(iters) == 0:
                 break
 
+            # Now that we are consuming data, start the commit thread
+            if self.commit_timer and not self.commit_timer.is_active:
+                self.commit_timer.start()
+
             for partition, it in iters.items():
                 try:
                     yield it.next()

kafka/util.py

@@ -86,14 +86,17 @@ def __init__(self, t, fn):
         self.timer = None
         self.t = t
         self.fn = fn
+        self.is_active = False
 
     def start(self):
         if self.timer is not None:
             self.timer.cancel()
 
         self.timer = Timer(self.t / 1000., self.fn)
+        self.is_active = True
         self.timer.start()
 
     def stop(self):
         self.timer.cancel()
         self.timer = None
+        self.is_active = False