Add more cleanup in consumer.stop()

mahendra · mahendra · commit c54a2edbaec1 · 2013-06-28T14:54:11.000+05:30
diff --git a/kafka/consumer.py b/kafka/consumer.py
@@ -159,6 +159,11 @@ def _auto_commit(self):
         if self.count_since_commit > self.auto_commit_every_n:
             self.commit()
 
+    def stop(self):
+        if self.commit_timer is not None:
+            self.commit_timer.stop()
+            self.commit()
+
     def pending(self, partitions=None):
         """
         Gets the pending message count
@@ -226,11 +231,6 @@ def provide_partition_info(self):
         """
         self.partition_info = True
 
-    def stop(self):
-        if self.commit_timer is not None:
-            self.commit_timer.stop()
-            self.commit()
-
     def seek(self, offset, whence):
         """
         Alter the current offset in the consumer, similar to fseek
@@ -510,6 +510,8 @@ def stop(self):
             proc.join()
             proc.terminate()
 
+        super(MultiProcessConsumer, self).stop()
+
     def __iter__(self):
         """
         Iterator to consume the messages available on this consumer
